Créez votre Data Lake

Dans ce chapitre on va aborder une notion simple mais importante d'Avro qui est l'évolution des schémas de données. Dans une application qui produit des données, la question n'est pas de savoir si le schéma des données va changer, mais quand. Les applications évoluent avec les besoins, les mises à jour techniques, bref la vie quoi. Il vaut donc mieux être prêt à gérer le changement.

Ajout d'un champ

Pour illustrer notre propos, reprenons le jeu de données géographique décrit dans les chapitres suivants. Pour mémoire, et si ce n'est pas encore fait, vous pouvez le télécharger en exécutant le script paris.py.

Ce jeu de données est constitué de fichiers XML qui contiennent, entre autres, des objets<node>. Nous avons sérialisé cesnodesen suivant le schéma suivant :

node.avsc:

{
    "namespace": "openclassrooms.openstreetmap",
    "type": "record",
    "name": "Node",
    "fields": [
        {"name": "id", "type": "long"},
        {"name": "longitude", "type": "float"},
        {"name": "latitude", "type": "float"},
        {"name": "username", "type": "string"},
    ]
}

Voici le script de sérialisation :

#! /usr/bin/env python3
# serialize.py
import json
import xml.etree.ElementTree as ET
import sys

import fastavro

# Read command line arguments
src = sys.argv[1]
dst = sys.argv[2]

# Read schema
schema = json.load(open("node.avsc"))

tree = ET.parse(open(src))
nodes = []
for node in tree.iterfind("node"):
    id = int(node.get("id"))
    longitude = float(node.get("lon"))
    latitude = float(node.get("lat"))
    username = node.get("user")
    nodes.append({
        "id": id,
        "longitude": longitude,
        "latitude": latitude,
        "username": username
    })
    print(id, longitude, latitude, username)

with open(dst, "wb") as avro_file:
    fastavro.writer(avro_file, schema, nodes)

On l'utilise pour sérialiser les données en provenance de2.364182,48.878215,2.365182,48.879215.osm:

$ python serialize.py ./data/paris/raw/2.364182,48.878215,2.365182,48.879215.osm ./nodes.avro

Le script de désérialisation, quant à lui, est très simple :

#! /usr/bin/env python
# deserialize.py
import json
import os
import sys

import fastavro

src = sys.argv[1]
schema = json.load(open(os.path.join(os.path.dirname(__file__), "node.avsc")))
with open(src, 'rb') as avro_file:
    reader = fastavro.reader(avro_file, reader_schema=schema)
    for node in reader:
        print(node)

On peut l'utiliser pour vérifier le contenu de notre fichiernodes.avro:

$ python deserialize.py ./nodes.avro
...
{'longitude': 2.3649117946624756, 'id': 735171905, 'username': 'RicB', 'latitude': 48.87907028198242}
...

En examinant les données brutes d'origine (contenue dans les fichiers*.osm), on remarque que certains<node>ont des<tag>qu'on n'a pas sérialisés. Par exemple, dans2.364182,48.878215,2.365182,48.879215.osm:

<node id="735171905" visible="true" version="5" changeset="25792149" timestamp="2014-10-01T15:26:49Z" user="RicB" uid="2303909" lat="48.8790715" lon="2.3649117">
  <tag k="amenity" v="restaurant"/>
  <tag k="name" v="L'École Buissonnière"/>
</node>

Comme on peut le voir, le node 735171905 possède deux tags ; chacun de ces tags a une cléket une valeurv. Voyant que nous avons oublié de sérialiser ces tags dans notre premier schéma, nous décidons de les y ajouter. Pour cela nous modifions le schéma :

{
    "namespace": "openclassrooms.openstreetmap",
    "type": "record",
    "name": "Node",
    "fields": [
        {"name": "id", "type": "long"},
        {"name": "longitude", "type": "float"},
        {"name": "latitude", "type": "float"},
        {"name": "username", "type": "string"},
        {"name": "tags", "type": {"type": "map", "values": "string"}}
    ]
}

Nos nodes sérialisés à l'aide de notre script vont désormais avoir un attributtagsqui est unemapcontenant des valeurs de typestring. Mais les nodes qui ont déjà été sérialisés ne pourront pas être désérialisés à l'aide de notre script. En effet, notre scriptdeserialize.pydéclenche maintenant une erreur :

$ python deserialize.py ./nodes.avro
Traceback (most recent call last):
  File "deserialize-simple.py", line 13, in <module>
    for node in reader:
  File "fastavro/_reader.py", line 511, in _iter_avro (fastavro/_reader.c:11517)
  File "fastavro/_reader.py", line 447, in fastavro._reader.read_data (fastavro/_reader.c:9910)
  File "fastavro/_reader.py", line 405, in fastavro._reader.read_record (fastavro/_reader.c:9572)
fastavro._reader.SchemaResolutionError: No default value for tags

En ajoutant un champ à notre schéma, nous avons brisé la rétrocompatibilité de notre schéma : il n'est plus possible de lire les anciennes données en utilisant le nouveau schéma. Vous vous rappelez que les données stockées dans notre master dataset doivent servir de référence pour toutes les applications actuelles et à venir ? Il est donc crucial que l'on puisse les lire de manière pérenne.

Heureusement, il existe une solution, qui est suggérée par le message d'erreur. Lorsque le champtagsn'existe pas, il suffit de considérer qu'il est égal à{}, soit la table de hâchage vide. Pour cela, on utilisé la propriétédefaultdu champtags:

{"name": "tags", "type": {"type": "map", "values": "string"}, "default": {}}

Avec ce nouveau schéma, on peut à la fois lire les anciennes données, et lire et écrire les nouvelles données qui possèdent le champtags.

Avro supporte différents types de changements dans les schémas. Avro utilise pour cela des règles de résolution de schémas.

Résolutions de schémas

Les règles appliquées par Avro pour résoudre l'évolution de schémas sont listées dans la documentation officielle. On retient notamment que :

1. Unintpeut être converti enlong,float,double, mais pas l'inverse. De manière générale, tout type numérique peut être converti vers un type plus complexe :longversfloat,floatversdouble, etc.

2. L'ordre des champs peut varier arbitrairement.

3. En cas d'ajout de champ, une valeur par défaut est nécessaire.

4. Il n'est possible de modifier unenumqu'en rajoutant des nouvelles valeurs.

En fait, la plupart de ces règles sont relativement intuitives ; en gros, l'enjeu est de ne pas perdre d'information entre deux évolutions de schémas.

Jusque récemment, il n'était pas possible de renommer un champ lors d'une évolution de schéma avec fastavro. La spécification d'Avro ne mentionne pas qu'il soit possible de renommer un champ, mais elle mentionne que les alias peuvent être utilisés à cette fin. Cependant, l'utilisation des alias est une fonctionnalité optionnelle qui n'est pas présente dans toutes les implémentations d'Avro. En particulier, fastavro ne supportait pas les alias, comme mentionné dans ce ticket, mais cette fonctionnalité a été implémentée depuis ^^. Lorsqu'un champ "name1" est renommé en "name2", vous pouvez faire évoluer votre schéma tout en restant compatible avec vos anciennes données en écrivant :

{
  "type": "record",
  "name": "name2",
  "aliases": ["name1"],
  "fields" : [...]
}

Si jamais vous avez besoin de réaliser une évolution draconienne de vos schémas de données, vous n'allez pas avoir le choix que de reprendre votre master dataset pour le modifier. Ce genre de situation est vraiment à éviter, d'autant que la quantité de données impliquées peut être importante. Dans ce cas, n'oubliez pas de réaliser un snapshot de votre master dataset, comme on l'a vu avec HDFS.