Envoyer des travaux Spark à l’aide d’outils en ligne de commande

Cet exemple utilise l’application PySpark suivante. Il est enregistré sous la forme d’un fichier Python nommé parquet_etl_sample.py sur l’ordinateur local.

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Read clickstream_data from storage pool HDFS into a Spark data frame. Applies column renames.
df = spark.read.option("inferSchema", "true").csv('/securelake/landing/criteo/test.txt', sep='\t', 
    header=False).toDF("feat1","feat2","feat3","feat4","feat5","feat6","feat7","feat8",
    "feat9","feat10","feat11","feat12","feat13","catfeat1","catfeat2","catfeat3","catfeat4",
    "catfeat5","catfeat6","catfeat7","catfeat8","catfeat9","catfeat10","catfeat11","catfeat12",
    "catfeat13","catfeat14","catfeat15","catfeat16","catfeat17","catfeat18","catfeat19",
    "catfeat20","catfeat21","catfeat22","catfeat23","catfeat24","catfeat25","catfeat26")

# Print the data frame inferred schema
df.printSchema()

tot_rows = df.count()
print("Number of rows:", tot_rows)

# Drop the managed table
spark.sql("DROP TABLE dl_clickstream")

# Write data frame to HDFS managed table by using optimized Delta Lake table format
df.write.format("parquet").mode("overwrite").saveAsTable("dl_clickstream")

print("Sample ETL pipeline completed")

Copier l’application PySpark dans HDFS

Stockez l’application dans HDFS afin que le cluster puisse y accéder pour l’exécution. En guise de meilleure pratique, standardisez et régissez les emplacements d’application au sein du cluster pour simplifier l’administration.

Dans cet exemple d’utilisation, toutes les applications de pipeline ETL sont stockées sur le chemin hdfs :/apps/ETL-Pipelines . L’exemple d’application est stocké sur hdfs :/apps/ETL-Pipelines/parquet_etl_sample.py.

Exécutez la commande suivante pour charger parquet_etl_sample.py à partir de l’ordinateur de développement local ou de préproduction sur le cluster HDFS.

azdata bdc hdfs cp --from-path parquet_etl_sample.py  --to-path "hdfs:/apps/ETL-Pipelines/parquet_etl_sample.py"

Cet exemple utilise une application Spark écrite dans Scala Spark.

import org.apache.spark.sql.SparkSession

object ParquetETLSample {
    def main(args: Array[String]) {
        val spark = SparkSession.builder.getOrCreate()
        
        val df = spark.read.
            option("inferSchema", "true").
            option("header", "false").
            option("delimiter", "\t").
            csv("/securelake/landing/criteo/test.txt").
            toDF("feat1","feat2","feat3","feat4","feat5","feat6","feat7","feat8","feat9","feat10","feat11","feat12","feat13","catfeat1","catfeat2","catfeat3","catfeat4","catfeat5","catfeat6","catfeat7","catfeat8","catfeat9","catfeat10","catfeat11","catfeat12","catfeat13","catfeat14","catfeat15","catfeat16","catfeat17","catfeat18","catfeat19","catfeat20","catfeat21","catfeat22","catfeat23","catfeat24","catfeat25","catfeat26")
        
        val tot_rows = df.count()
        println(s"Number of rows: $tot_rows")

        spark.sql("DROP TABLE dl_clickstream")

        df.write.format("parquet").mode("overwrite").saveAsTable("dl_clickstream")

        println("Sample ETL pipeline completed")
        
        spark.stop()
    }
}

Regrouper et copier l’application Spark dans HDFS

La documentation Spark recommande de créer un fichier JAR d’assembly (ou bundle) qui contient votre application et toutes les dépendances. Cette étape est nécessaire pour envoyer le bundle d’applications au cluster pour l’exécution.

La configuration d’un environnement de développement Scala Spark complet dépasse la portée de cet article. Pour plus d’informations, consultez la documentation Spark pour la création d’applications autonomes.

Cet exemple suppose qu’un bundle JAR d’application nommé parquet-etl-sample.jar est compilé et disponible. Exécutez la commande suivante pour charger le bundle à partir de l’ordinateur de développement local ou de préproduction sur le cluster HDFS.

azdata bdc hdfs cp --from-path parquet-etl-sample.jar  --to-path "hdfs:/apps/ETL-Pipelines/parquet-etl-sample.jar"

Cet exemple utilise Spark SQL pour la logique d’ingestion. Il utilise des tables et des vues pour fournir une approche centrée sur SQL pour ETL.

DROP VIEW IF EXISTS etl_clickstream;

CREATE TEMPORARY VIEW etl_clickstream
USING CSV
OPTIONS (path "/securelake/landing/criteo/test.txt", header "false", delimiter "\t", mode "FAILFAST");

DROP TABLE IF EXISTS dl_clickstream;

CREATE TABLE dl_clickstream (
    feat1 integer,
    feat2 integer,
    feat3 integer,
    feat4 integer,
    feat5 integer,
    feat6 integer,
    feat7 integer,
    feat8 integer,
    feat9 integer,
    feat10 integer,
    feat11 integer,
    feat12 integer,
    feat13 integer,
    catfeat1 string,
    catfeat2 string,
    catfeat3 string,
    catfeat4 string,
    catfeat5 string,
    catfeat6 string,
    catfeat7 string,
    catfeat8 string,
    catfeat9 string,
    catfeat10 string,
    catfeat11 string,
    catfeat12 string,
    catfeat13 string,
    catfeat14 string,
    catfeat15 string,
    catfeat16 string,
    catfeat17 string,
    catfeat18 string,
    catfeat19 string,
    catfeat20 string,
    catfeat21 string,
    catfeat22 string,
    catfeat23 string,
    catfeat24 string,
    catfeat25 string,
    catfeat26 string
) 
USING PARQUET
AS SELECT * FROM etl_clickstream;

Copier l’application Spark SQL dans HDFS

Exécutez la commande suivante pour charger le fichier parquet-etl-sample.sql à partir de l’ordinateur de développement local ou de préproduction sur le cluster HDFS.

azdata bdc hdfs cp --from-path parquet-etl-sample.sql --to-path "hdfs:/apps/ETL-Pipelines/parquet-etl-sample.sql"

La azdata commande exécute l’application à l’aide de paramètres couramment spécifiés. Pour obtenir les options de paramètre complètes pour azdata bdc spark batch create, consultez azdata bdc spark.

Cette application nécessite le spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation paramètre de configuration. Par conséquent, la commande utilise l’option --config . Cette configuration montre comment transmettre des configurations à la session Spark.

Vous pouvez utiliser l’option --config pour spécifier plusieurs paramètres de configuration. Vous pouvez également les spécifier à l’intérieur de la session d’application en définissant la configuration dans l’objet SparkSession .

azdata bdc spark batch create -f hdfs:/apps/ETL-Pipelines/parquet_etl_sample.py \
--config '{"spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation":"true"}' \
-n MyETLPipelinePySpark --executor-count 2 --executor-cores 2 --executor-memory 1664m

La curl commande exécute l’application à l’aide de Livy. Remplacez USER, PASSWORDet LIVY_ENDPOINT reflètez votre environnement.

curl -k -u <USER>:<PASSWORD> -X POST <LIVY_ENDPOINT>/batches \
-H 'Content-Type: application/json; charset=utf-8' \
--data-binary @- << EOF
{
    "file": "/apps/ETL-Pipelines/parquet_etl_sample.py",
    "name": "MyETLPipelinePySpark",
    "numExecutors": 2,
    "executorCores": 2,
    "executorMemory": "1664m",
    "conf": {
        "spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation":"true"
    }
}
EOF

La azdata commande exécute l’application à l’aide de paramètres couramment spécifiés. Pour obtenir les options de paramètre complètes pour azdata bdc spark batch create, consultez azdata bdc spark.

L’application nécessite le spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation paramètre de configuration. Par conséquent, la commande utilise l’option --config . Cette configuration montre comment transmettre des configurations à la session Spark.

Vous pouvez utiliser l’option --config pour spécifier plusieurs paramètres de configuration. Vous pouvez également les spécifier à l’intérieur de la session d’application en définissant la configuration dans l’objet SparkSession .

azdata bdc spark batch create -f hdfs:/apps/ETL-Pipelines/parquet-etl-sample.jar \
--class "ParquetETLSample" \
--config '{"spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation":"true"}' \
-n MyETLPipeline --executor-count 2 --executor-cores 2 --executor-memory 1664m

La curl commande exécute l’application à l’aide de Livy. Remplacez USER, PASSWORDet LIVY_ENDPOINT reflètez votre environnement.

curl -k -u <USER>:<PASSWORD> -X POST <LIVY_ENDPOINT>/batches \
-H 'Content-Type: application/json; charset=utf-8' \
--data-binary @- << EOF
{
    "file": "/apps/ETL-Pipelines/parquet-etl-sample.jar",
    "class": "ParquetETLSample",
    "name": "MyETLPipeline",
    "numExecutors": 2,
    "executorCores": 2,
    "executorMemory": "1664m",
    "conf": {
        "spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation":"true"
    }
}
EOF

La azdata commande exécute l’application à l’aide de paramètres couramment spécifiés. Pour obtenir les options de paramètre complètes pour azdata bdc spark batch create, consultez azdata bdc spark.

Comme l’exemple PySpark, cette application nécessite également le spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation paramètre de configuration. Par conséquent, la commande utilise l’option --config . Cette configuration montre comment transmettre des configurations à la session Spark.

Vous pouvez utiliser l’option --config pour spécifier plusieurs paramètres de configuration. Vous pouvez également les spécifier à l’intérieur de la session d’application en définissant la configuration dans l’objet SparkSession .

azdata bdc spark batch create -f hdfs:/apps/ETL-Pipelines/parquet_etl_sample.sql \
--config '{"spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation":"true"}' \
-n MyETLPipelineSQL --executor-count 2 --executor-cores 2 --executor-memory 1664m

La curl commande exécute l’application à l’aide de Livy. Remplacez USER, PASSWORDet LIVY_ENDPOINT reflètez votre environnement.

curl -k -u <USER>:<PASSWORD> -X POST <LIVY_ENDPOINT>/batches \
-H 'Content-Type: application/json; charset=utf-8' \
--data-binary @- << EOF
{
    "file": "/apps/ETL-Pipelines/parquet_etl_sample.sql",
    "name": "MyETLPipelineSQL",
    "numExecutors": 2,
    "executorCores": 2,
    "executorMemory": "1664m",
    "conf": {
        "spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation":"true"
    }
}
EOF