Inviare processi Spark usando strumenti da riga di comando

In questo esempio viene usata l'applicazione PySpark seguente. Viene salvato come file Python denominato parquet_etl_sample.py nel computer locale.

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")

Copiare l'applicazione PySpark in HDFS

Archiviare l'applicazione in HDFS in modo che il cluster possa accedervi per l'esecuzione. Come procedura consigliata, standardizzare e gestire i percorsi delle applicazioni all'interno del cluster per semplificare l'amministrazione.

In questo caso d'uso di esempio tutte le applicazioni della pipeline ETL vengono archiviate nel percorso hdfs:/apps/ETL-Pipelines. L'applicazione di esempio viene archiviata in hdfs:/apps/ETL-Pipelines/parquet_etl_sample.py.

Eseguire il comando seguente per caricare parquet_etl_sample.py dal computer di sviluppo locale o di gestione temporanea nel cluster HDFS.

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

In questo esempio si usa un'applicazione Spark scritta in 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()
    }
}

Creare un bundle e copiare l'applicazione Spark in HDFS

Nella documentazione di Spark si consiglia di creare un file JAR di assembly (o pacchetto) contenente l'applicazione e tutte le dipendenze. Questo passaggio è necessario per inviare il pacchetto dell'applicazione al cluster per l'esecuzione.

La configurazione di un ambiente di sviluppo Scala Spark completo esula dall'ambito di questo articolo. Per altre informazioni, vedere la documentazione di Spark relativa alla creazione di applicazioni autonome.

In questo esempio si presuppone che un pacchetto JAR dell'applicazione denominato parquet-etl-sample.jar sia compilato e disponibile. Eseguire il comando seguente per caricare il pacchetto dal computer di sviluppo locale o di gestione temporanea nel cluster HDFS.

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

In questo esempio su usa Spark SQL per la logica di inserimento. Usa tabelle e viste per fornire all'applicazione ETL un approccio incentrato su SQL.

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;

Copiare l'applicazione Spark SQL in HDFS

Eseguire il comando seguente per caricare il file parquet-etl-sample.sql dal computer di sviluppo locale o di gestione temporanea nel cluster HDFS.

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

Il comando azdata esegue l'applicazione usando parametri comunemente specificati. Per tutte le opzioni dei parametri relativi a azdata bdc spark batch create, vedere azdata bdc spark.

Questa applicazione richiede il parametro di configurazione spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation. Il comando usa quindi l'opzione --config. Questa configurazione illustra come passare le configurazioni alla sessione Spark.

È possibile usare l'opzione --config per specificare più parametri di configurazione. È anche possibile specificarli all'interno della sessione dell'applicazione impostando la configurazione nell'oggetto 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

Il comando curl esegue l'applicazione con Livy. Sostituire USER, PASSWORD e LIVY_ENDPOINT in base all'ambiente usato.

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

Il comando azdata esegue l'applicazione usando parametri comunemente specificati. Per tutte le opzioni dei parametri relativi a azdata bdc spark batch create, vedere azdata bdc spark.

L'applicazione richiede il parametro di configurazione spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation. Il comando usa quindi l'opzione --config. Questa configurazione illustra come passare le configurazioni alla sessione Spark.

È possibile usare l'opzione --config per specificare più parametri di configurazione. È anche possibile specificarli all'interno della sessione dell'applicazione impostando la configurazione nell'oggetto 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

Il comando curl esegue l'applicazione con Livy. Sostituire USER, PASSWORD e LIVY_ENDPOINT in base all'ambiente usato.

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

Il comando azdata esegue l'applicazione usando parametri comunemente specificati. Per tutte le opzioni dei parametri relativi a azdata bdc spark batch create, vedere azdata bdc spark.

Analogamente all'esempio PySpark, questa applicazione richiede anche il parametro di configurazione spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation. Il comando usa quindi l'opzione --config. Questa configurazione illustra come passare le configurazioni alla sessione Spark.

È possibile usare l'opzione --config per specificare più parametri di configurazione. È anche possibile specificarli all'interno della sessione dell'applicazione impostando la configurazione nell'oggetto 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

Il comando curl esegue l'applicazione con Livy. Sostituire USER, PASSWORD e LIVY_ENDPOINT in base all'ambiente usato.

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