Schema Merging com Spark e Java

Schema Merging é uma forma de evoluir schemas adicionando novas colunas através da junção de DataFrames. Imagine que você possui diferentes arquivos parquets com diferentes schemas, e que exista a necessidade em criar um novo schema a partir de todas as colunas destes variados parquets. Podemos resolver este problema em uma simples linha de código, conforme será mostrado a seguir.

A seguir vamos criar arquivos parquets com diferentes schemas através de arquivos JSON e em seguida, faremos o merge destes arquivos transformando em um único schema consolidado.

Arquivos JSON que usaremos como DataSource:

Arquivo user.json

{"id":1, "login": "Jonh", "age": 24}
{"id":2, "login": "Lucas", "age": 24}
{"id":3, "login": "Miesha", "age": 25}
{"id":4, "login": "Suzie", "age": 22}

Arquivo address.json

{"id":1, "city": "Los Angeles", "country": "USA"}
{"id":2, "city": "New York", "country": "USA"}
{"id":3, "city": "San Louis Obispo", "country": "USA"}

Criando o SparkSession

SparkConf sparkConf = new SparkConf();
sparkConf.setMaster("local[1]");
sparkConf.setAppName("app");

SparkSession sparkSession =
        new SparkSession.Builder()
                .config(sparkConf)
                .getOrCreate();

Criando os DataFrames

Dataset<Row> dfUser =
        sparkSession.read().json("user.json");

Dataset<Row> dfAddress =
        sparkSession.read().json("address.json");

Gerando os Parquets para cada DataFrame

No código a seguir, estamos criando os parquets no diretório data/table/ para a partição chamada partition de valor 1

dfUser.write().parquet("data/table/partition=1");

No código a seguir, estamos criando os parquets no diretório data/table/ para a partição chamada partition de valor 2

dfAddress.write().parquet("data/table/partition=2");

Agora, criaremos um novo DataFrame com base nos parquets criados anteriormente executando o Schema Merging

Dataset<Row> dfMerge = sparkSession
        .read().option("mergeSchema", true)
        .parquet("data/table");

O Schema Merging é executado no seguinte trecho

option("mergeSchema", true)

Por fim, podemos ver o novo schema mergeado através do mergeSchema

dfMerge.printSchema();

Resultado

root

|-- age: long (nullable = true)

|-- id: long (nullable = true)

|-- login: string (nullable = true)

|-- city: string (nullable = true)

|-- country: string (nullable = true)

|-- partition: integer (nullable = true)

Veja que a partir dos DataFrames dfUser e dfAddress foi criado um novo schema consolidando as colunas entre eles.

Código completo

SparkConf sparkConf = new SparkConf();
sparkConf.setMaster("local[1]");
sparkConf.setAppName("app");

SparkSession sparkSession =
        new SparkSession.Builder()
                .config(sparkConf)
                .getOrCreate();

Dataset<Row> dfUser =
        sparkSession.read().json("user.json");

dfUser.write().parquet("data/table/partition=1");

Dataset<Row> dfAddress =
        sparkSession.read().json("address.json");

dfAddress.write().parquet("data/table/partition=2");

Dataset<Row> dfMerge = sparkSession
        .read().option("mergeSchema", true)
        .parquet("data/table");

dfMerge.printSchema();

Curtiu? Espero que sim, até mais!