Smart_open: Data loss while writing avro file to s3 compatible storage

[vinuthna91]

Hi,

I am converting a csv file into avro and writing to s3 compliant storage.I see that schema file(.avsc) is written properly. However, there is data loss while writing to .avro file.
Below is snippet of my code

## Code
import smart_open
from boto.compat import urlsplit, six
import boto
import boto.s3.connection

import avro.schema
from avro.datafile import  DataFileWriter 
from avro.io import  DatumWriter

import pandas as pn
import os,sys

FilePath = 's3a://mybucket/vinuthnav/csv/file1.csv' #path on s3

splitInputDir = urlsplit(FilePath, allow_fragments=False)

inConn = boto.connect_s3(
    aws_access_key_id = access_key_id,
    aws_secret_access_key = secret_access_key,
    port=int(port),
    host = hostname,
    is_secure=False,
    calling_format = boto.s3.connection.OrdinaryCallingFormat(),
    )
#get bucket
inbucket = inConn.get_bucket(splitInputDir.netloc)
#read in the csv file
kr = inbucket.get_key(splitInputDir.path)
with smart_open.smart_open(kr, 'r') as fin:
    xa = pn.read_csv(fin, header=1, error_bad_lines = False).fillna('NA')

rowCount, columnCount = xa.shape #check if data frame is empty, if it is, don't write outp
if rowCount == 0:
    ##do nothing
    print '>> [NOTE] empty file'


else:
    #generate avro schema and data

    dataFile = os.path.join(os.path.basename(FileName), os.path.splitext(FileName)[0]+".avro")
    schemaFile = os.path.join(os.path.basename(FileName), os.path.splitext(FileName)[0]+".avsc")

    kwd = inbucket.get_key(urlsplit(dataFile, allow_fragments=False).path, validate=False)
    schema = gen_schema(xa.columns)

    with smart_open.smart_open(kwd, 'wb') as foutd: 

        dictRes = xa.to_dict(orient='records')
        writer = DataFileWriter(foutd, DatumWriter(), schema)
        for ll, row in enumerate(dictRes):
            writer.append(row)

[mpenkov]

Thank you for providing example source. Could you please describe what the problem is in detail? An example of the data loss, as well as how you detected it, would be very helpful.

[vinuthna91]

I converted the avro file to Json and read the number of lines. I expect the number of lines in Json to match the number of lines in csv and it didn't match.

[mpenkov]

Have you confirmed that this problem still persists if you use local storage instead of smart_open?

[vinuthna91]

Yes, this problem doesn't exist if I use local storage . However file on local storage was not written through smart_open

[mpenkov]

From your example source, I can see that you're reading the CSV file with smart_open too. Is the CSV file being read correctly? In other words, does the problem still persist if the CSV file is loaded from local storage?

[vinuthna91]

Yes file is being read correctly. I counted number of lines in the dataframe i loaded . i also counted the datumWriter count. It matches but after writing to avro on s3 and reading it i see there is data loss. This confirms that problem is with writing binary data to s3

[mpenkov]

What version of smart_open are you using? Have you observed a similar problem with older versions?

[vinuthna91]

smart_open-1.3.5 with anaconda version of python 2.7

[vinuthna91]

I haven't tried with other versions

[mpenkov]

OK, thank you for providing detailed information about this bug. We will investigate.

[vinuthna91]

thanks

[vinuthna91]

Hi, do you have any update?

[mpenkov]

Sorry, no, I have not looked into this yet. Is this urgent for you?

[vinuthna91]

Yes, this is a blocker for us . We are also searching some alternative to smart open. Haven't found any yet. At least a work around asap would be great

[mpenkov]

I think the workaround is to write the avro file to local storage first, and upload to S3 when it is complete.

I've started investigating the issue, but cannot reproduce the problem because your code sample is incomplete: the gen_schema function is missing. Could you please look at this file:

https://github.com/mpenkov/smart_open/blob/209/integration-tests/test_209.py

and update it so that it reproduces your problem?

[vinuthna91]

Thanks for investigating the issue. writing to local storage is not an option for us because of storage problem and latency it involves
I updated the file. Please try it and let me know.

[mpenkov]

Where is the updated file?

[vinuthna91]

I have updated the code in test_209.py you provided. It showed pushed

[mpenkov]

I think your push failed. Here’s the file in your repo: https://github.com/vinuthna91/smart_open/blob/209/integration-tests/test_209.py

It’s unchanged from my version.

If you cant work it out, please paste the code here as a comment.

[vinuthna91]

I changed it, I am sorry for delayed response. I was out of station.

https://github.com/vinuthna91/smart_open/blob/209/integration-tests/test_209.py

def gen_schema(paramNames):
    paramNamesLen = len(paramNames)


    dataName = 'schema'
    avroSchemaOut = "{\n\t\"type\":     \"record\", \"name\": \"%s\", \"namespace\": \"com.sandisk.bigdata\", \n \t\"fields\": [" %(dataName)  


    if paramNamesLen==0:
        #no parameters, no schema file generation
        avroSchemaOut = ''

    else:
        #generate file
        for ii in range(paramNamesLen):
            typeString = "[\"%s\", \"null\"]" %('String')
            schemaString = "{ \"name\":\"%s\", \"type\":%s, \"default\":null}" % (paramNames[ii], typeString)
            if ii == 0:
                avroSchemaOut += schemaString + ',\n'
            elif ii <len(paramNames)-1:
                avroSchemaOut += "\t\t\t" + schemaString + ',\n'
            else:
                avroSchemaOut += "\t\t\t" + schemaString + '\n'
        avroSchemaOut += "\n \t\t\t]\n}"

    return avroSchemaOut

[mpenkov]

I pulled your changes and made some updates. Unfortunately, the code crashes with an error unrelated to smart_open:

Traceback (most recent call last): File "integration-tests/test_209.py", line 61, in <module> writer.append(row) File "/Users/misha/envs/smartopen27/lib/python2.7/site-packages/avro/datafile.py", line 196, in append self.datum_writer.write(datum, self.buffer_encoder) File "/Users/misha/envs/smartopen27/lib/python2.7/site-packages/avro/io.py", line 768, in write if not validate(self.writers_schema, datum): File "/Users/misha/envs/smartopen27/lib/python2.7/site-packages/avro/io.py", line 103, in validate schema_type = expected_schema.type AttributeError: 'str' object has no attribute 'type'

Please have a look and update the code. The updated code is here: https://github.com/mpenkov/smart_open/blob/209/integration-tests/test_209.py

[vinuthna91]

Updated code is here:
https://github.com/mpenkov/smart_open/pull/1/commits/3eaf46f0ca48d23b50f15921bdb16fe812ed17e2

[mpenkov]

Thank you for the update. Unfortunately, I still cannot reproduce your problem because of errors in avro.

avro.io.AvroTypeException: The datum {'93493318071517': 93493318071467, 'MULEY FANATIC FOUNDATION OF WYOMING INC': 'KALAMAZOO COMMUNITY FOUNDATION', '201713189349307151': 201713189349307146, 'EFILE': 'EFILE', '201612': 201612, '453578215': 383333202, '15109264': 15109263, '1/10/2018 1:03:25 PM': '1/10/2018 1:03:23 PM', '990': '990'} is not an example of the schema {
  "namespace": "com.sandisk.bigdata",
  "type": "record",
  "name": "schema",
  "fields": [
    {
      "default": null,
      "type": [
        "null",
        "string"
      ],
      "name": "15109264"
    },
    {
      "default": null,
      "type": [
        "null",
        "string"
      ],
      "name": "EFILE"
    },

Can you please make sure the test runs and reproduces your actual problem?

[vinuthna91]

Can you give the csv you used?

[mpenkov]

The CSV gets downloaded as part of the actual test. Please see the source
file.
On Thu, Jul 26, 2018 at 20:46 vinuthna91 [e-mail] wrote:

Can you give the csv you used?

—
You are receiving this because you commented.

Reply to this email directly, view it on GitHub
https://github.com/RaRe-Technologies/smart_open/issues/209#issuecomment-408070182,
or mute the thread
https://github.com/notifications/unsubscribe-auth/ABDOVNa4S6NuvZly7GhcHDkbWl-zmRhYks5uKawQgaJpZM4VPAWw
.

[vinuthna91]

Could you please check now. I updated the code.
https://github.com/vinuthna91/smart_open/blob/209/integration-tests/test_209.py

[vinuthna91]

Were you able to try it? Please confirm that the code is working now

[mpenkov]

Yes, the code is working now. I was able to use it to reproduce the problem. I also simplified it a bit to rule out errors in the actual test: https://github.com/mpenkov/smart_open/blob/209/integration-tests/test_209.py

I'll take it from here. Thanks for your help in reproducing the problem.

[mpenkov]

$ SO_NUMROWS=10000 SO_S3_URL=s3://redacted/test_smart_open python integration-tests/test_209.py
download: s3://redacted/test_smart_open/issue_209/out.avro to ./remote.avro
Binary files local.avro and remote.avro differ

[mpenkov]

The two avro files are:

1. local.avro: written straight to local storage
2. remote.avro: written to S3 storage via smart_open, and then retrieved via the CLI

They should be the same, but they are different for some reason, suggesting there is a problem with smart_open.

[mpenkov]

Some more details:

(smart_open2.7) misha@cabron:~/git/smart_open$ SO_NUMROWS=477 SO_S3_URL=s3://redacted/smart_open_test/ python integration-tests/test_209.py 
CRITICAL:smart_open.s3:wrote 8192 bytes (8192 total)
CRITICAL:smart_open.s3:wrote 8192 bytes (16384 total)
CRITICAL:smart_open.s3:wrote 8192 bytes (24576 total)
CRITICAL:smart_open.s3:wrote 8192 bytes (32768 total)
CRITICAL:smart_open.s3:wrote 8192 bytes (40960 total)
CRITICAL:smart_open.s3:wrote 8192 bytes (49152 total)
CRITICAL:smart_open.s3:wrote 8192 bytes (57344 total)
CRITICAL:smart_open.s3:wrote 7473 bytes (64817 total)
download: s3://redacted/smart_open_test//issue_209/out.avro to ./remote.avro
sanity check OK
CRITICAL:smart_open.s3:wrote 4 bytes (4 total)
CRITICAL:smart_open.s3:wrote 1 bytes (5 total)
CRITICAL:smart_open.s3:wrote 1 bytes (6 total)
CRITICAL:smart_open.s3:wrote 11 bytes (17 total)
CRITICAL:smart_open.s3:wrote 1 bytes (18 total)
CRITICAL:smart_open.s3:wrote 1 bytes (19 total)
CRITICAL:smart_open.s3:wrote 680 bytes (699 total)
CRITICAL:smart_open.s3:wrote 1 bytes (700 total)
CRITICAL:smart_open.s3:wrote 10 bytes (710 total)
CRITICAL:smart_open.s3:wrote 1 bytes (711 total)
CRITICAL:smart_open.s3:wrote 4 bytes (715 total)
CRITICAL:smart_open.s3:wrote 1 bytes (716 total)
CRITICAL:smart_open.s3:wrote 16 bytes (732 total)
CRITICAL:smart_open.s3:wrote 1 bytes (733 total)
CRITICAL:smart_open.s3:wrote 1 bytes (734 total)
CRITICAL:smart_open.s3:wrote 1 bytes (735 total)
CRITICAL:smart_open.s3:wrote 1 bytes (736 total)
CRITICAL:smart_open.s3:wrote 1 bytes (737 total)
CRITICAL:smart_open.s3:wrote 64064 bytes (64801 total)
CRITICAL:smart_open.s3:wrote 16 bytes (64817 total)
download: s3://redacted/smart_open_test//issue_209/out.avro to ./remote.avro
Binary files local.avro and remote.avro differ
test NG
(smart_open2.7) misha@cabron:~/git/smart_open$ hexdump local.avro > local.hex
(smart_open2.7) misha@cabron:~/git/smart_open$ hexdump remote.avro > remote.hex
(smart_open2.7) misha@cabron:~/git/smart_open$ diff --suppress-common-lines --side-by-side remote.hex local.hex 
00002c0 632e 646f 6365 6e08 6c75 006c c515 49a0               | 00002c0 632e 646f 6365 6e08 6c75 006c db3c 70da
00002d0 35ce aed7 8e65 0591 4a65 5a8e 07ba e980               | 00002d0 4162 c706 555f 8eb3 987a 395b 07ba e980
000fd20 1535 a0c5 ce49 d735 65ae 918e 6505 8e4a               | 000fd20 3c35 dadb 6270 0641 5fc7 b355 7a8e 5b98
000fd30 005a                                                  | 000fd30 0039 

The same number of bytes get written in both cases, but there seems to be some corruption happening from bytes 02c0 to 02e0 (704 to 736 in decimal); and fd20 to fd40 (64800 to 64832 in decimal).

Interesting points:

• The first badness includes several write boundaries.
• The second badness is in the last 32 bytes on the file. It includes a boundary between writes.

[mpenkov]

@vinuthna91 I've had a look at the issue. It appears avro is sending _slightly_ different data when writing locally compared to when writing to S3 via smart_open. Have a look at this branch.

I've also added logging to the library in the critical places. This logging is important for demonstrating the problem.

Now, when we run the test, we get this:

(smart_open2.7) misha@cabron:~/git/smart_open$ SO_NUMROWS=477 SO_S3_URL=s3://redacted/smart_open_test/ python integration-tests/test_209.py 
CRITICAL:root:writing to <_io.BufferedWriter name='local.avro'>                                                                             
CRITICAL:smart_open.smart_open_lib:wrote 10 bytes (710 total)                                                                               
CRITICAL:smart_open.smart_open_lib:6176 726f 2e63 6f64 6563                                                                                 
CRITICAL:smart_open.smart_open_lib:wrote 1 bytes (711 total)                                                                                
CRITICAL:smart_open.smart_open_lib:08                                                                                                       
CRITICAL:smart_open.smart_open_lib:wrote 4 bytes (715 total)
CRITICAL:smart_open.smart_open_lib:6e75 6c6c
CRITICAL:smart_open.smart_open_lib:wrote 1 bytes (716 total)
CRITICAL:smart_open.smart_open_lib:00
CRITICAL:smart_open.smart_open_lib:wrote 16 bytes (732 total)
CRITICAL:smart_open.smart_open_lib:83a8 3a59 9a9c 0ba4 b495 9802 bd78 ef59
CRITICAL:smart_open.smart_open_lib:wrote 1 bytes (733 total)
CRITICAL:smart_open.smart_open_lib:ba
CRITICAL:smart_open.smart_open_lib:wrote 1 bytes (734 total)
CRITICAL:smart_open.smart_open_lib:07
CRITICAL:smart_open.smart_open_lib:wrote 1 bytes (735 total)
CRITICAL:smart_open.smart_open_lib:80
CRITICAL:root:writing to <smart_open.s3.BufferedOutputBase object at 0x7fdc65025310>
CRITICAL:smart_open.s3:wrote 10 bytes (710 total)
CRITICAL:smart_open.s3:6176 726f 2e63 6f64 6563
CRITICAL:smart_open.s3:wrote 1 bytes (711 total)
CRITICAL:smart_open.s3:08
CRITICAL:smart_open.s3:wrote 4 bytes (715 total)
CRITICAL:smart_open.s3:6e75 6c6c
CRITICAL:smart_open.s3:wrote 1 bytes (716 total)
CRITICAL:smart_open.s3:00
CRITICAL:smart_open.s3:wrote 16 bytes (732 total)
CRITICAL:smart_open.s3:557a 9c7f 1ee2 88eb 41bf 0dfb 880d b30e
CRITICAL:smart_open.s3:wrote 1 bytes (733 total)
CRITICAL:smart_open.s3:ba
CRITICAL:smart_open.s3:wrote 1 bytes (734 total)
CRITICAL:smart_open.s3:07
CRITICAL:smart_open.s3:wrote 1 bytes (735 total)
CRITICAL:smart_open.s3:80
download: s3://redacted/smart_open_test//issue_209/out.avro to ./remote.avro
Binary files local.avro and remote.avro differ
test NG
(smart_open2.7) misha@cabron:~/git/smart_open$

There are two separate writes happening here:

1. avro writes to local storage via smart_open
2. avro writes to S3 via smart_open

Have a look at the bytes that avro is passing to smart_open. In the first case:

CRITICAL:smart_open.smart_open_lib:wrote 16 bytes (732 total)
CRITICAL:smart_open.smart_open_lib:83a8 3a59 9a9c 0ba4 b495 9802 bd78 ef59

and in the second case:

CRITICAL:smart_open.s3:wrote 16 bytes (732 total)
CRITICAL:smart_open.s3:557a 9c7f 1ee2 88eb 41bf 0dfb 880d b30e

In both cases, those bytes are being passed to smart_open directly from avro. The bytes that avro is passing to smart_open are different in both cases. This explains why the result between writing locally and writing to S3 is different.

Therefore, I think this may be a problem in avro, not smart_open. @menshikh-iv Please have a look and double-check my reasoning. It's also possible I messed something up in the code, so please have a look at that, too.

[vinuthna91]

Avro takes foutd as argument which is file writer. For local storage it is just fopen argument, while in s3 case, it is generated from smart open isn't it? Do you think avro cannot generate correct data for smart open key?

[vinuthna91]

Could you suggest any other way of writing avro files to s3 using python instead of using avro datum writer?

[mpenkov]

For local storage it is just fopen argument, while in s3 case, it is generated from smart open isn't it?

Yes.

Do you think avro cannot generate correct data for smart open key?

I'm not 100% sure, but that's the only plausible explanation I can offer, given the test results above. To be 100% sure of what's going on, you'd have to debug avro, confirm that it's sending different data, and look for a reason why.

Could you suggest any other way of writing avro files to s3 using python instead of using avro datum writer?

Sorry, I'm entirely unfamiliar with avro.

I have a follow-up question for you. How do you convert the avro files to JSON, so that you can see the effect of data corruption?

[vinuthna91]

I have a follow-up question for you. How do you convert the avro files to JSON, so that you can see the effect of data corruption?

I use a Java jar to do this. Command below-
java -jar avro-tools-1.8.2.jar tojson > jsonfilename.json

[scottbelden]

As the maintainer of fastavro I wanted to try this and see if any problem exists when using that library.

I tried running the reproduction script and it's failing with something in smart_open. Can you help me understand what the problem is (I'm completely unfamiliar with smart_open and have never used it).

CRITICAL:root:writing to <_io.BufferedWriter name='local.avro'>
Traceback (most recent call last):
  File "test.py", line 112, in <module>
    with smart_open.smart_open(output_url, 'wb') as foutd:
  File "/home/sbelden/.virtualenvs/smartopen/lib/python2.7/site-packages/smart_open/smart_open_lib.py", line 231, in smart_open
    binary, filename = _open_binary_stream(uri, binary_mode, **kw)
  File "/home/sbelden/.virtualenvs/smartopen/lib/python2.7/site-packages/smart_open/smart_open_lib.py", line 321, in _open_binary_stream
    return _s3_open_uri(parsed_uri, mode, **kw), filename
  File "/home/sbelden/.virtualenvs/smartopen/lib/python2.7/site-packages/smart_open/smart_open_lib.py", line 383, in _s3_open_uri
    return smart_open_s3.open(parsed_uri.bucket_id, parsed_uri.key_id, mode, **kwargs)
  File "/home/sbelden/.virtualenvs/smartopen/lib/python2.7/site-packages/smart_open/s3.py", line 73, in open
    fileobj = BufferedOutputBase(bucket_id, key_id, min_part_size=s3_min_part_size, **kwargs)
  File "/home/sbelden/.virtualenvs/smartopen/lib/python2.7/site-packages/smart_open/s3.py", line 365, in __init__
    raise ValueError('the bucket %r does not exist, or is forbidden for access' % bucket)
ValueError: the bucket 'redacted' does not exist, or is forbidden for access

[mpenkov]

@scottbelden Thank you for looking into this.

S3 is a cloud-based storage solution from Amazon Web Services. S3 stores data in collections called buckets. In your case, you're trying to write to a bucket that you don't have write access to, as the error at the end of the stack trace is trying to tell you.

To help you solve the problem, I've added the option to mock out S3. This way, you can reproduce the original problem without having write access to an S3 bucket. Please do a pull from my 209 branch, and then run:

$ SO_ENABLE_MOCKS=1 SO_NUMROWS=477 SO_S3_URL=s3://dummy python integration-tests/test_209.py

[scottbelden]

Thanks.

Doing a binary diff of the files to determine if they are the "same" won't work. This is because part of the avro specification includes a random sync interval in the file and so even if you were comparing the local.avro across two invocations it would be different.

However, if we modify the avro library to always use the same sync interval then we can compare them. So by changing avro/datafile.py to return b"1234567890abcdef" you can ensure the same sync interval is used. Then when I re-ran the 209.py script I got a test OK

[mpenkov]

OK, then we need to make the tests more robust. Can fastavro convert avro files to JSON? I'd rather not have to include java in the test to do the same thing.

I have a follow-up question for you. How do you convert the avro files to JSON, so that you can see the effect of data corruption?

I use a Java jar to do this. Command below-
java -jar avro-tools-1.8.2.jar tojson > jsonfilename.json

[scottbelden]

Converting to JSON is something I'm actually working on in my free time at the moment, but it doesn't quite work yet. However, you can use the regular avro library to read the file and spit out the list of records and compare those.

To read the records you can do this:

def read_avro(foutd):
    reader = avro.datafile.DataFileReader(foutd, avro.io.DatumReader())
    return list(reader)

And then you could add some check like you have with the binary diff:

with open('local.avro', 'rb') as fout:
    local_records = read_avro(fout)

with open('remote.avro', 'rb') as fout:
    remote_records = read_avro(fout)

if local_records == remote_records:
    print('Same')
else:
    print('Different')

When I do that I get:

Binary files local.avro and remote.avro differ
test NG

[mpenkov]

@scottbelden Thank you for providing insight about avro. I've updated the test.

@vinuthna91 Please see the updated test. Everything passes, so we cannot reproduce your error anymore. We get the same result with and without mocking.

Please try to reproduce the error using the updated test code.

CRITICAL:root:writing to <open file 'local.avro', mode 'wb' at 0x7f87d5c60150>
CRITICAL:root:writing to <_io.BufferedWriter name='local-so.avro'>
sanity check OK
CRITICAL:root:output_url: 's3://dummy/issue_209/out.avro' bucket_name: 'dummy' key_name: 'issue_209/out.avro'
CRITICAL:root:writing to <smart_open.s3.BufferedOutputBase object at 0x7f87d21bfbd0>
test OK
binary check OK
(smart_open2.7) misha@cabron:~/git/smart_open$ SO_S3_URL=s3://dbi2-ue2/smart_open_test python integration-tests/test_209.py
CRITICAL:root:writing to <open file 'local.avro', mode 'wb' at 0x7fa6594cc150>
CRITICAL:root:writing to <_io.BufferedWriter name='local-so.avro'>
sanity check OK
CRITICAL:root:output_url: 's3://redacted/smart_open_test/issue_209/out.avro' bucket_name: 'dbi2-ue2' key_name: 'smart_open_test/issue_209/out.avro'
CRITICAL:root:writing to <smart_open.s3.BufferedOutputBase object at 0x7fa65596d310>
test OK
binary check OK
(smart_open2.7) misha@cabron:~/git/smart_open$

[vinuthna91]

so, fix is to set the avro sync marker to same value to avoid data loss?
@mock.patch('avro.datafile.DataFileWriter.generate_sync_marker', mock.Mock(return_value=b'0123456789abcdef'))

[mpenkov]

No, that line is there so that the sync marker does not interfere with the binary check. You may remove it during your tests, but if you do so, be sure to disable the binary check at the end as well.

The current test results show that there is no data loss. Please confirm using your own data and let me know if you get a different result.

[vinuthna91]

Ok, I will check and let you know

[vinuthna91]

I checked on my dataset. local.avro and remote.avro matched but total number of records in avro file and csv file did not match even by reading with read_avro function in your code.

[vinuthna91]

number of records read by avro should be same as number of data lines

[mpenkov]

If local.avro and remote.avro matched, the problem isn't in smart_open.

[vinuthna91]

@mpenkov thanks for the help. I will debug more to see what is the issue in the script.

[vinuthna91]

@mpenkov I have an observation:

while i read csv from local storage, convert it to avro
local.avro and remote.avro do not differ.

However, i read csv from s3, convert it to avro
local.avro and remote.avro differ.

[mpenkov]

Without seeing your actual test case (source code or at least shell commands), it's hard for me to comment.

I ran an additional test to ensure that there is no data loss when _reading_ files from S3:

(smart_open2.7) misha@cabron:~/git/smart_open$ python integration-tests/s3_cp.py s3://irs-form-990/index_2018.csv index_2018_smartopen.csv
(smart_open2.7) misha@cabron:~/git/smart_open$ aws s3 cp s3://irs-form-990/index_2018.csv index_2018.csv 
download: s3://irs-form-990/index_2018.csv to ./index_2018.csv     
(smart_open2.7) misha@cabron:~/git/smart_open$ diff index_2018_smartopen.csv index_2018.csv 
(smart_open2.7) misha@cabron:~/git/smart_open$

I downloaded the CSV file using 1) smart_open 2) AWS CLI and compared the two files. They were identical, so there is no data loss in the reading case, either.

[vinuthna91]

I have used same code provided by you except that i used my local data set. My csv file size is only 90KB

On line number 72, I have added #writer.close(). When writer.close() is uncommented, writing local.avro to local storage is complete and there is no data loss. However, writing to s3 storage throws an error AttributeError: 'S3OpenWrite' object has no attribute 'flush'.
Here local.avro and remote.avro differ

When you comment this( writer.close() is not used), local.avro and remote.avro both are same but, there is data loss in both the files when compared to csv data.
That is required to write avro completely.

[scottbelden]

I don't know how pandas works, but if you look at the data variable which is the result of running the csv file through pandas you'll notice that it only has [477 rows x 9 columns]. The csv file has over 172,000 rows, so it seems like some data is getting truncated there.

[scottbelden]

Oh, the truncation is just because of the SO_NUMROWS=477. By changing that to 200000 all of the lines are processed

[vinuthna91]

i don't see truncation in number of rows in pandas data frame. i see it only in the avro file. I am counting number of records written to local avro file when we use writer.close() and when we do not use.

[scottbelden]

When I read in all the data using SO_NUMROWS=200000 and then print the resulting number or records I see that local.avro has 172396 while remote.avro has 172389

[vinuthna91]

when writer.close() is called it internally calls a flush() method that Flushes the current state of the file, including metadata. The error i am getting is 'S3OpenWrite' object has no attribute 'flush'. because of which metadata is being lost.

[scottbelden]

I don't see that exception from the test script, but yeah, that would make sense.

[vinuthna91]

@scottbelden that is exactly the data loss i am talking about.

If you uncomment writer.close() in line#72, you should see the exception while writing to avro file on s3

[scottbelden]

I'm not sure which script has writer.close() on line 72...

[vinuthna91]

I made the change to test script. If you do not see the change, you need to add writer.close() to the write_avro function definition:

def write_avro(foutd):
schema = avro.schema.parse(avroSchemaOut)
dictRes = data.to_dict(orient='records')
writer = avro.datafile.DataFileWriter(foutd, avro.io.DatumWriter(), schema)
for ll, row in enumerate(dictRes):
writer.append(row)
writer.close()

[mpenkov]

@vinuthna91 Why are you closing the writer manually? The writer is a context manager. It will take care of closing and cleaning up when the context exits (at the end of the "with" block).

OK, I think see what's going on here. I'll look at this again on the weekend.

[vinuthna91]

Thanks!

[vinuthna91]

could you find anything?

[mpenkov]

OK, I think see what's going on here. I'll look at this again on the weekend.

It's still Thursday. Weekend starts in two days.

[vinuthna91]

Okay.

[mpenkov]

OK, it's the weekend, and I've had a look at it. Time to clear things up.

The "data loss" referred to in this ticket does not come from smart_open. It comes from misusing avro. You need to either:

1. Close the writer using writer.close (as @vinuthna91 mentioned a few posts ago)
2. Use the writer as a context manager.

Both work identically. If you don't call them, avro keeps some data in its internal buffers, and never writes it to the output file.

However, if you do the above while using smart_open, avro ends up calling BufferedOutputWriter.close twice. The first time succeeds, but the second time fails due to a bug in that method. I've added a test and fixed that bug.

Strangely, I haven't been able to reproduce the errors related to the absence of a flush method. @vinuthna91, can you reproduce the error with the new code?

[vinuthna91]

@mpenkov Thanks for looking into it.

unfortunately, i used the same code and it still throws me error about absence of flush method.
Error below-

CRITICAL:root:writing to <open file 'local.avro', mode 'wb' at 0x7fdb98674ae0>
CRITICAL:root:writing to <open file 'local-so.avro', mode 'wb' at 0x7fdb98674b70>
CRITICAL:root:writing to <open file 'local-nomanual.avro', mode 'wb' at 0x7fdb98674ae0>
CRITICAL:root:writing to <smart_open.smart_open_lib.S3OpenWrite object at 0x7fdb98624c90>
CRITICAL:root:writing to <smart_open.smart_open_lib.S3OpenWrite object at 0x7fdb985ef790>
ERROR:smart_open.smart_open_lib:encountered error while terminating multipart upload; attempting cancel
Traceback (most recent call last):
  File "test_smartopenavro.py", line 129, in <module>
    write_avro_context_manager(foutd)
  File "test_smartopenavro.py", line 68, in write_avro_context_manager
    writer_contextManager.append(row)
  File "/usr/anaconda/lib/python2.7/site-packages/avro/datafile.py", line 131, in __exit__
    self.close()
  File "/usr/anaconda/lib/python2.7/site-packages/avro/datafile.py", line 219, in close
    self.flush()
  File "/usr/anaconda/lib/python2.7/site-packages/avro/datafile.py", line 215, in flush
    self.writer.flush()
AttributeError: 'S3OpenWrite' object has no attribute 'flush'
Traceback (most recent call last):
  File "test_smartopenavro.py", line 129, in <module>
    write_avro_context_manager(foutd)
  File "test_smartopenavro.py", line 68, in write_avro_context_manager
    writer_contextManager.append(row)
  File "/usr/anaconda/lib/python2.7/site-packages/avro/datafile.py", line 131, in __exit__
    self.close()
  File "/usr/anaconda/lib/python2.7/site-packages/avro/datafile.py", line 219, in close
    self.flush()
  File "/usr/anaconda/lib/python2.7/site-packages/avro/datafile.py", line 215, in flush
    self.writer.flush()
AttributeError: 'S3OpenWrite' object has no attribute 'flush'

Since, i get the error, I also added another method which doesnt have manual close. that how i am trying to write now-

def write_avro_no_manual_close(foutd):
    schema = avro.schema.parse(avroSchemaOut)
    dictRes = data.to_dict(orient='records')
    writer = avro.datafile.DataFileWriter(foutd, avro.io.DatumWriter(), schema)
    for ll, row in enumerate(dictRes):
        writer.append(row)
    #writer.close()

def write_avro_context_manager(foutd):
    schema = avro.schema.parse(avroSchemaOut)
    dictRes = data.to_dict(orient='records')
    with avro.datafile.DataFileWriter(foutd, avro.io.DatumWriter(), schema) as writer:
        for ll, row in enumerate(dictRes):
            writer.append(row)

@mock.patch('avro.datafile.DataFileWriter.generate_sync_marker', mock.Mock(return_value=b'0123456789abcdef'))
def write_avro_manual_close(foutd):
    schema = avro.schema.parse(avroSchemaOut)
    dictRes = data.to_dict(orient='records')
    writer = avro.datafile.DataFileWriter(foutd, avro.io.DatumWriter(), schema)
    for ll, row in enumerate(dictRes):
        writer.append(row)
    writer.close()

with open('local.avro', 'wb') as foutd:
    logging.critical('writing to %r', foutd)
    write_avro_context_manager(foutd)

with open('local-so.avro', 'wb') as foutd:
    logging.critical('writing to %r', foutd)
    write_avro_manual_close(foutd)

with open('local-nomanual.avro', 'wb') as foutd:
    logging.critical('writing to %r', foutd)
    write_avro_no_manual_close(foutd)

I clearly see that local.avro, local-so.avro are identical but not identical to local-nomanual.avro.

I am pasting the code i am using also for your reference-

import os
import os.path as P

import avro.io
import avro.datafile
import pandas as pn
import smart_open
import six

import subprocess
import warnings
import boto
from boto.compat import urlsplit, six
import boto.s3.connection

import logging
import json

###### Added by me
access_key_id=access_key_id
secret_access_key =secret_access_key
port=port
hostname=hostname
##########

with warnings.catch_warnings():
    warnings.simplefilter('ignore')
    import pandas as pn

logging.basicConfig(level=logging.ERROR)

if six.PY3:
    assert False, 'this code only runs on Py2.7'

_S3_URL = 's3://bucket-for-testing/user/vinuthna'
assert _S3_URL is not None, 'please set the SO_S3_URL environment variable'

'''
_NUMROWS = os.environ.get('SO_NUMROWS')
if _NUMROWS is not None:
    _NUMROWS = int(_NUMROWS)


'''
def gen_schema(data):
    schema = {
        'type': 'record', 'name': 'data', 'namespace': 'namespace',
        'fields': [
            {'name': field, 'type': ['null', 'string'], 'default': None}
            for field in data.columns
        ]
    }
    return json.dumps(schema, indent=4)

def write_avro_nomanualclose(foutd):
    schema = avro.schema.parse(avroSchemaOut)
    dictRes = data.to_dict(orient='records')
    writer_nomanual = avro.datafile.DataFileWriter(foutd, avro.io.DatumWriter(), schema)
    for ll, row in enumerate(dictRes):
        writer_nomanual.append(row)
    #writer_manual.close()

def write_avro_context_manager(foutd):
    schema = avro.schema.parse(avroSchemaOut)
    dictRes = data.to_dict(orient='records')
    with avro.datafile.DataFileWriter(foutd, avro.io.DatumWriter(), schema) as writer_contextManager:
        for ll, row in enumerate(dictRes):
            writer_contextManager.append(row)

def write_avro_manual_close(foutd):
    schema = avro.schema.parse(avroSchemaOut)
    dictRes = data.to_dict(orient='records')
    writer_manual = avro.datafile.DataFileWriter(foutd, avro.io.DatumWriter(), schema)
    for ll, row in enumerate(dictRes):
        writer_manual.append(row)
    writer_manual.close()

inputFilePath = P.join(_S3_URL,'test.csv')

splitInputDir = urlsplit(inputFilePath, allow_fragments=False)

#logging.info(">>> establishing connection for inputDir")
#establish connection manuelly for customized options
inConn = boto.connect_s3(
    aws_access_key_id = access_key_id,
    aws_secret_access_key = secret_access_key,
    port=int(port),
    host = hostname,
    is_secure=False,
    calling_format = boto.s3.connection.OrdinaryCallingFormat(),
    )
#get bucket
inbucket = inConn.get_bucket(splitInputDir.netloc)
# read in the csv file
kr = inbucket.get_key(splitInputDir.path)

assert kr is not None, 'File not present'

with smart_open.smart_open(kr, 'r') as fin:
    data = pn.read_csv(fin, header=1, error_bad_lines=False,dtype='str').fillna('NA')

num_csv_rows = len(data.index)
avroSchemaOut = gen_schema(data)

#No dataloss
with open('local.avro', 'wb') as foutd:
    logging.critical('writing to %r', foutd)
    write_avro_context_manager(foutd)

###No dataloss
with open('local-so.avro', 'wb') as foutd:
    logging.critical('writing to %r', foutd)
    write_avro_manual_close(foutd)

## Data loss observed
with open('local-nomanual.avro', 'wb') as foutd:
    logging.critical('writing to %r', foutd)
    write_avro_nomanualclose(foutd)

##Below works
outputfilepath = _S3_URL + '/nomanual.avro'
splitoutdir = urlsplit(outputfilepath, allow_fragments=False)
kw = inbucket.get_key(splitoutdir.path,validate=False)
with smart_open.smart_open(kw, 'wb') as foutd:
    logging.critical('writing to %r', foutd)
    write_avro_nomanualclose(foutd)

### This throws error
outputfilepath = _S3_URL + '/outcontext.avro'
splitoutdir = urlsplit(outputfilepath, allow_fragments=False)
kw = inbucket.get_key(splitoutdir.path,validate=False)
with smart_open.smart_open(kw, 'wb') as foutd:
    logging.critical('writing to %r', foutd)
    write_avro_context_manager(foutd)

[vinuthna91]

Also, in your test code, try reading csv file from s3 because my test case is reading a csv from s3 -> converting to avro -> writing back avro to s3

[mpenkov]

Whoa... Please format your code next time. Besides, most of that code is not relevant, because it demonstrates facts that we've already established (if you don't close the Avro writer, you'll lose some data).

Also, in your test code, try reading csv file from s3 because my test case is reading a csv from s3 -> converting to avro -> writing back avro to s3

Thank you for giving me details about your test case, I don't think this is relevant to the problem.

What version of avro are you using?

[vinuthna91]

Sure, I will format it. I am using version 2.0

[mpenkov]

That doesn't sound right.

(smart_open2.7) misha@cabron:~/git/smart_open$ pip freeze | grep avro
avro==1.8.2
(smart_open2.7) misha@cabron:~/git/smart_open$ pip install --upgrade avro
Requirement already up-to-date: avro in /home/misha/envs/smart_open2.7/lib/python2.7/site-packages (1.8.2)
(smart_open2.7) misha@cabron:~/git/smart_open$ 

The newest version in pip is 1.8.2. The current version on apache.org is also 1.8.2.

[vinuthna91]

sorry it is 1.8.0. I misinterpreted it

[mpenkov]

I tried to reproduce the problem with 1.8.0 but couldn't.

In the immediate future, you could try mocking out the flush method:

# completely untested code!
with smart_open.smart_open(s3_url, 'wb') as fout:
    fout.flush = lambda: None
    write_avro(data, fout)

I'll look at adding flush support to the API.

[vinuthna91]

Thanks. It says object is read only

Traceback (most recent call last):
File "test_smartopenavro.py", line 119, in
write_avro_nomanualclose(foutd)
File "test_smartopenavro.py", line 59, in write_avro_nomanualclose
foutd.flush = lambda: None
AttributeError: 'file' object attribute 'flush' is read-only

[mpenkov]

Sounds like you're overriding flush for the wrong object.

AttributeError: 'file' object attribute 'flush' is read-only

This means that flush already exists for that particular object.

[vinuthna91]

Got it. Its working now . Thanks a lot for the work around :)

[mpenkov]

Closed via https://github.com/RaRe-Technologies/smart_open/pull/212