Dataframes wijzigen en opslaan
Apache Spark biedt het dataframeobject als de primaire structuur voor het werken met gegevens. U kunt dataframes gebruiken om gegevens op te vragen en te transformeren en de resultaten in een data lake te behouden. Als u gegevens wilt laden in een dataframe, gebruikt u de functie spark.read , waarbij u de bestandsindeling, het pad en eventueel het schema van de gegevens opgeeft dat moet worden gelezen. Met de volgende code worden bijvoorbeeld gegevens uit alle CSV-bestanden in de map Orders in een dataframe met de naam order_details geladen en worden vervolgens de eerste vijf records weergegeven.
order_details = spark.read.csv('/orders/*.csv', header=True, inferSchema=True)
display(order_details.limit(5))
De gegevensstructuur transformeren
Nadat u de brongegevens in een dataframe hebt geladen, kunt u de methoden en Spark-functies van het dataframe-object gebruiken om deze te transformeren. Typische bewerkingen in een dataframe zijn onder andere:
- Rijen en kolommen filteren
- De naam van kolommen wijzigen
- Nieuwe kolommen maken, vaak afgeleid van bestaande kolommen
- Null- of andere waarden vervangen
In het volgende voorbeeld gebruikt de code de split functie om de waarden in de kolom CustomerName te scheiden in twee nieuwe kolommen met de naam FirstName en LastName. Vervolgens wordt de drop methode gebruikt om de oorspronkelijke kolom CustomerName te verwijderen.
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))
U kunt de volledige kracht van de Spark SQL-bibliotheek gebruiken om de gegevens te transformeren door rijen te filteren, af te leiden, te verwijderen, de naam van kolommen te wijzigen en andere vereiste gegevenswijzigingen toe te passen.
De getransformeerde gegevens opslaan
Nadat uw dataFrame zich in de vereiste structuur bevindt, kunt u de resultaten opslaan in een ondersteunde indeling in uw Data Lake.
In het volgende codevoorbeeld wordt het dataFrame opgeslagen in een Parquet-bestand in de Data Lake, waarbij een bestaand bestand met dezelfde naam wordt vervangen.
transformed_df.write.mode("overwrite").parquet('/transformed_data/orders.parquet')
print ("Transformed data saved!")
Notitie
De Parquet-indeling heeft doorgaans de voorkeur voor gegevensbestanden die u gaat gebruiken voor verdere analyse of opname in een analytische opslag. Parquet is een zeer efficiënte indeling die wordt ondersteund door de meeste grootschalige systemen voor gegevensanalyse. Soms is het zelfs mogelijk dat uw gegevenstransformatie vereist is om gegevens te converteren van een andere indeling (zoals CSV) naar Parquet.