Modificar e salvar dataframes
O Apache Spark fornece o objeto dataframe como a estrutura primária para trabalhar com os dados. Você pode usar dataframes para consultar e transformar dados e persistir os resultados em um data lake. Para carregar os dados em um dataframe, use a função spark.read, especificando o formato do arquivo, o caminho e, opcionalmente, o esquema dos dados a serem lidos. Por exemplo, o código a seguir carrega dados de todos os arquivos .csv da pasta orders em um dataframe chamado order_details e exibe os cinco primeiros registros.
order_details = spark.read.csv('/orders/*.csv', header=True, inferSchema=True)
display(order_details.limit(5))
Transformar a estrutura de dados
Depois de carregar os dados de origem em um dataframe, use os métodos do objeto dataframe e as funções do Spark para transformá-los. As operações típicas em um dataframe incluem:
- Filtragem de linhas e colunas
- Renomeação de colunas
- Criação de colunas, muitas vezes derivadas das existentes
- Substituição de valores nulos ou outros
No exemplo a seguir, o código usa a função split para separar os valores na coluna CustomerName em duas novas colunas chamadas FirstName e LastName. Em seguida, ele usa o método drop para excluir a coluna CustomerName original.
from pyspark.sql.functions import split, col
# Create the new FirstName and LastName fields
transformed_df = order_details.withColumn("FirstName", split(col("CustomerName"), " ").getItem(0)).withColumn("LastName", split(col("CustomerName"), " ").getItem(1))
# Remove the CustomerName field
transformed_df = transformed_df.drop("CustomerName")
display(transformed_df.limit(5))
Use todo o potencial da biblioteca Spark SQL para transformar os dados filtrando linhas, derivando, removendo, renomeando colunas e aplicando outras modificações de dados necessárias.
Salvar os dados transformados
Depois que o dataFrame estiver na estrutura necessária, você poderá salvar os resultados em um formato com suporte no data lake.
O exemplo de código a seguir salva o dataFrame em um arquivo Parquet no data lake, substituindo os arquivos existentes com o mesmo nome.
transformed_df.write.mode("overwrite").parquet('/transformed_data/orders.parquet')
print ("Transformed data saved!")
Observação
Normalmente, o formato Parquet é preferencial para os arquivos de dados que você usará para análise ou ingestão posterior em um repositório analítico. O Parquet é um formato muito eficiente que é compatível com a maioria dos sistemas de análise de dados em grande escala. Na verdade, às vezes, seu requisito de transformação de dados pode ser apenas converter dados de outro formato (como CSV) em Parquet.