Convert avro schema

From raju

Dummy

Problem Description

We are given an avro file where a field is stored as a string even though the underlying values are all integers. This is sub optimal as it increases the file size. The goal here is to create a new avro file where that field is stored as an integer.

Sample Input

Generate a simple avro file by running the write_avro.py given in the appendix

    $python write_avro.py
    wrote weather.avro
    

Here the sample field is stored as strings even though the underlying values are all integers.

    $fastavro weather.avro
    {"station": "New York City", "sample": "3628", "temp": 42.29999923706055}
    {"station": "San Jose", "sample": "0389", "temp": 67.4000015258789}
    {"station": "Hyderabad", "sample": "3379", "temp": 104.69999694824219}
    {"station": "New Delhi", "sample": "5478", "temp": 98.75}
    
    $fastavro --schema weather.avro
    {
     "type": "record",
     "name": "com.company.group.objects.avro.weather.SampleData",
     "fields": [
      {
       "name": "station",
       "type": "string"
      },
      {
       "name": "sample",
       "type": "string"
      },
      {
       "name": "temp",
       "type": "float"
      }
     ],
     "__fastavro_parsed": true
    }
    

Solution

Convert the schema and the underlying data by running convert_schema.py given in the appendix.

    $python convert_schema.py
    reading weather.avro
    wrote weather_cmpct.avro with new schema
    

The new avro file, weather_cmpct.avro, stores the sample field as integers.

    $fastavro weather_cmpct.avro
    {"station": "New York City", "sample": 3628, "temp": 42.29999923706055}
    {"station": "San Jose", "sample": 389, "temp": 67.4000015258789}
    {"station": "Hyderabad", "sample": 3379, "temp": 104.69999694824219}
    {"station": "New Delhi", "sample": 5478, "temp": 98.75}
    
    $fastavro --schema weather_cmpct.avro
    {
     "type": "record",
     "name": "com.company.group.objects.avro.weather.SampleData",
     "fields": [
      {
       "name": "station",
       "type": "string"
      },
      {
       "name": "sample",
       "type": "int"
      },
      {
       "name": "temp",
       "type": "float"
      }
     ],
     "__fastavro_parsed": true
    }
    

As expected, the new avro file is smaller compared to the old (365 bytes vs. 380 bytes).

    $du -b weather.avro weather_cmpct.avro
    380     weather.avro
    365     weather_cmpct.avro
    

Appendix

write_avro.py

    $cat write_avro.py
    # Write a simple avro file
    
    from fastavro import writer, parse_schema
    
    def get_schema():
        schema = {
            'type': 'record',
            'name': 'SampleData',
            'namespace': 'com.company.group.objects.avro.weather',
            'fields': [
                {'name': 'station', 'type': 'string'},
                {'name': 'sample', 'type': 'string'},
                {'name': 'temp', 'type': 'float'},
            ],
        }
        parsed_schema = parse_schema(schema)
        return parsed_schema
    
    def write_avro():
        schema = get_schema()
        records = [
            {'station': 'New York City', 'sample': '3628', 'temp': 42.3},
            {'station': 'San Jose', 'sample': '0389', 'temp': 67.4},
            {'station': 'Hyderabad', 'sample': '3379', 'temp': 104.7},
            {'station': 'New Delhi', 'sample': '5478', 'temp': 98.75},
        ]
        file_path = 'weather.avro'
        with open(file_path, 'wb') as out:
            writer(out, schema, records)
        print('wrote', file_path)
    
    if __name__ == '__main__':
        write_avro()
    

convert_schema.py

    $cat convert_schema.py
    # In weather.avro, the sample field is stored as string. Change it to int.
    
    from fastavro import reader, writer, parse_schema
    
    def get_new_schema():
        schema = {
            'type': 'record',
            'name': 'SampleData',
            'namespace': 'com.company.group.objects.avro.weather',
            'fields': [
                {'name': 'station', 'type': 'string'},
                {'name': 'sample', 'type': 'int'},
                {'name': 'temp', 'type': 'float'},
            ],
        }
        parsed_schema = parse_schema(schema)
        return parsed_schema
    
    def convert_schema():
        schema = get_new_schema()
    
        old_file = 'weather.avro'
        print('reading', old_file)
        with open(old_file, 'rb') as fin:
            records = [r for r in reader(fin)]
    
        for r in records:
            r['sample'] = int(r['sample'])
    
        new_file = 'weather_cmpct.avro'
        with open(new_file, 'wb') as fout:
            writer(fout, schema, records)
        print('wrote', new_file, 'with new schema')
    
    if __name__ == '__main__':
        convert_schema()