U /es#@sddlZddlZddlmZmZmZddlmZddlmZ ddl m Z dZ dZ dd Zd d Zd d ZddZd ddZddZddZd!ddZddZddZdS)"N)OpenFileget_fs_token_paths open_files) read_block)tokenize)HighLevelGraphsObjcCsd|d}t|}|d@}d}|d@dkrRt|d}||d@|>O}|d7}q|d?|d@ AS)z"variable-length, zig-zag encoding.r)readord)focbnshiftr1/tmp/pip-unpacked-wheel-dbjnr7gq/dask/bag/avro.py read_longs   rcCst|}||S)z+a long followed by that many bytes of data.)rr )rsizerrr read_bytessrcCs|tttkstdi}d|i}t|}|dkr8qXt|D]}t|t|q@q&|t|d<||d<| d||d|d<|S)zExtract an avro file's header fo: file-like This should be in bytes mode, e.g., io.BytesIO Returns dict representing the header Parameters ---------- fo: file-like zMagic avro bytes missingmetarsync header_size head_bytes) r lenMAGICAssertionErrorrranger SYNC_SIZEtellseek)rroutZn_keys_rrr read_header!s     r&c Cs8t|||d}t|}W5QRX||d}||fS)z*Open a file just to read its head and size compressionr)rr&info)fspathr(fheadrrrr open_head?sr.csddlm}m}ddlm}ddlm}|dd|p8i}|dk rbt|d|d \}} |t|fd d | D} t | \} } |t g} g}| D]6}t t d||}|gt |}| |||qg} t | | || D]t\}}}d }t|d t|||||fdd |D}fdd t |||D}| |q|| St|fdi|}|tfdd |D}||SdS)aRead set of avro files Use this with arbitrary nested avro schemas. Please refer to the fastavro documentation for its capabilities: https://github.com/fastavro/fastavro Parameters ---------- urlpath: string or list Absolute or relative filepath, URL (may include protocols like ``s3://``), or globstring pointing to data. blocksize: int or None Size of chunks in bytes. If None, there will be no chunking and each file will become one partition. storage_options: dict or None passed to backend file-system compression: str or None Compression format of the targe(s), like 'gzip'. Should only be used with blocksize=None. r)computedelayed) from_delayedimport_requiredfastavrozjszread_avro..rr'csg|]}d|dqS)z read-avro--r)r9o)tokenrrr;}sc s$g|]\}}}|||dqS))Z dask_key_namer)r9r=keyl)dreadr,r-rrr;~sr(csg|] }|qSrr)r9r)rArrr;s)Zdaskr0r1Zdask.bagr2 dask.utilsr4rr.zip read_chunklistr rappendr fs_tokenizeZukeyextendr read_file)Zurlpath blocksizer8r(r0r1r2r4Zfs_tokenpathsr$ZheadsZsizesoffsetslengthsrofflengthr+offset delimiterkeysvaluesfileschunksr)r(r:rAr,r*r-r>r read_avroGs^         rVc Csrddl}t|dr|j}n|j}|}t||||d}W5QRX|d}|ts\||}t|} t || S)zGet rows from raw bytes blockrN iter_avrorr) r5hasattrrWreaderr startswithrioBytesIOrE) ZfobjrNr@r-r5rYr,chunkrirrrrDs   rDc CsHddl}t|dr|j}n|j}|}t||W5QRSQRXdS)zGet rows from file-likerNrW)r5rXrWrYrE)rr5rYr,rrrrIs  rInull>Tc  sddlm} | ddt|p$i}t|df|jd|} dtjfdd t| D} t j | gd } t | j}|r|j f| d d | DS| Sd S)a Write bag to set of avro files The schema is a complex dictionary describing the data, see https://avro.apache.org/docs/1.8.2/gettingstartedpython.html#Defining+a+schema and https://fastavro.readthedocs.io/en/latest/writer.html . It's structure is as follows:: {'name': 'Test', 'namespace': 'Test', 'doc': 'Descriptive text', 'type': 'record', 'fields': [ {'name': 'a', 'type': 'int'}, ]} where the "name" field is required, but "namespace" and "doc" are optional descriptors; "type" must always be "record". The list of fields should have an entry for every key of the input records, and the types are like the primitive, complex or logical types of the Avro spec ( https://avro.apache.org/docs/1.8.2/spec.html ). Results in one avro file per input partition. Parameters ---------- b: dask.bag.Bag filename: list of str or str Filenames to write to. If a list, number must match the number of partitions. If a string, must include a glob character "*", which will be expanded using name_function schema: dict Avro schema dictionary, see above name_function: None or callable Expands integers into strings, see ``dask.bytes.utils.build_name_function`` storage_options: None or dict Extra key/value options to pass to the backend file-system codec: 'null', 'deflate', or 'snappy' Compression algorithm sync_interval: int Number of records to include in each block within a file metadata: None or dict Included in the file header compute: bool If True, files are written immediately, and function blocks. If False, returns delayed objects, which can be computed by the user where convenient. kwargs: passed to compute(), if compute=True Examples -------- >>> import dask.bag as db >>> b = db.from_sequence([{'name': 'Alice', 'value': 100}, ... {'name': 'Bob', 'value': 200}]) >>> schema = {'name': 'People', 'doc': "Set of people's scores", ... 'type': 'record', ... 'fields': [ ... {'name': 'name', 'type': 'string'}, ... {'name': 'value', 'type': 'int'}]} >>> b.to_avro('my-data.*.avro', schema) # doctest: +SKIP ['my-data.0.avro', 'my-data.1.avro'] rr3r5z:fastavro is a required dependency for using bag.to_avro().wb) name_functionnumzto-avro-c s.i|]&\}}|ftj|f|fqSr)_write_avro_partname)r9r^r,rcodecmetadatareschema sync_intervalrr s zto_avro..)Z dependenciescSsg|] }|jqSr)r+)r9r,rrrr;szto_avro..N)rBr4_verify_schemarZ npartitionsuuidZuuid4hex enumeraterZfrom_collectionstyper0Z to_delayed)rfilenamerirbr8rgrjrhr0kwargsr4rTZdskgraphr$rrfrto_avros4K    rtcCst|tstddD]}||kstd|q|ddksDtdt|dtsZtd|dD] }d |krvd|ksbtd |qbdS) NzSchema must be dictionary)rerpfieldszSchema missing '%s' fieldrprecordzSchema must be of type 'record'ruzFields entry must be a listrezField spec incomplete: %s) isinstancedictrrE)sfieldr,rrrrls rlc Cs0ddl}|}|||||||W5QRXdS)z1Create single avro file from list of dictionariesrN)r5writer)partr,rirgrjrhr5rrrrd%srd)r/NN)NNr_r`NT)r[rmZ fsspec.corerrrZ fsspec.utilsrrrGZdask.highlevelgraphrrr!rrr&r.rVrDrIrtrlrdrrrrs.     E p