filipelenfers

def dict_to_item(raw:dict):
    return {
            key: _dict_to_item_recurse(value)
            for key, value in raw.items()
        }


def _dict_to_item_recurse(raw):
    if isinstance(raw, dict):
        return {

filipelenfers

Avro serialization

There are 4 possible serialization format when using avro:

• Avro Json encoding
• Avro Data Serialization (https://avro.apache.org/docs/current/spec.html#Data+Serialization) Binary format with an header that contains the full schema, this is the format usually used when writing Avro files
• Avro Single Object Encoding (https://avro.apache.org/docs/current/spec.html#single_object_encoding) Binary format with an header with only the fingerprint/id of the schema, this it the format used by Kafka (see this
• Avro Binary Encoding (https://avro.apache.org/docs/current/spec.html#binary_encoding)

filipelenfers

import io
import snappy
import avro.schema
from avro.datafile import DataFileReader, DataFileWriter
from avro.io import DatumReader, DatumWriter, BinaryEncoder

#Generate avro bytes as a in-memory file, with the schema.
def generate_avro_bytes(schema,avro_dicts, codec='snappy'):
    bytes_writer = io.BytesIO()
    writer = DataFileWriter(bytes_writer, DatumWriter(), schema, codec=codec)

filipelenfers

# !!!THIS SCRIPT ASSUME HIVE COMPATIBLE PARTITIONS!!!
#
# When you get the following error:
# HIVE_PARTITION_SCHEMA_MISMATCH: There is a mismatch between the table and partition schemas. The types are incompatible and cannot be coerced.
# what happened is that some column type changed in the table, and partitions have they own version of the schema, 
# and the version that partitions have cached are not compatible with the table actual version on glue.
# We need to:
# 1. remove partitions
# 2. add partitions again
#