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!