Enviar trabalhos do Spark usando ferramentas de linha de comando

Este exemplo usa o aplicativo do PySpark a seguir. Ele é salvo como um arquivo Python chamado parquet_etl_sample.py no computador 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")

Copiar o aplicativo PySpark para HDFS

Armazene o aplicativo no HDFS de modo que o cluster possa acessá-lo para execução. Como melhor prática, padronize e controle os locais de aplicativo no cluster para simplificar a administração.

Neste caso de uso de exemplo, todos os aplicativos de pipeline ETL estão armazenados no caminho hdfs:/apps/ETL-Pipelines. O aplicativo de exemplo é armazenado em HDFS:/apps/ETL-pipelines/parquet_etl_sample. py.

Execute o comando a seguir para carregar o parquet_etl_sample.py do computador de preparo ou de desenvolvimento local no cluster HDFS.

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

Este exemplo usa um aplicativo Spark escrito no 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()
    }
}

Agrupar e copiar o aplicativo Spark no HDFS

A documentação do Spark recomenda a criação de um JAR do assembly (ou pacote) que contém seu aplicativo e todas as dependências. Essa etapa é necessária para enviar o pacote de aplicativos ao cluster para execução.

Configurar um ambiente de desenvolvimento do Scala Spark completo está além do escopo deste artigo. Para obter mais informações, confira a documentação do Spark para criar aplicativos independentes.

Este exemplo pressupõe que um pacote JAR do aplicativo chamado parquet-etl-sample.jar foi compilado e está disponível. Execute o comando a seguir para carregar o pacote do computador de preparo ou de desenvolvimento local no cluster HDFS.

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

Este exemplo usa o Spark SQL para a lógica de ingestão. Ele usa tabelas e exibições para fornecer uma abordagem centrada em SQL para 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;

Copiar o aplicativo Spark SQL no HDFS

Execute o comando a seguir para carregar o arquivo parquet-etl-sample.sql do computador de preparo ou de desenvolvimento local no cluster HDFS.

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

O comando azdata executa o aplicativo usando parâmetros comumente especificados. Para obter opções de parâmetro completas para azdata bdc spark batch create, consulte azdata bdc spark.

Esse aplicativo requer o parâmetro de configuração spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation. Portanto, o comando usa a opção --config. Essa configuração mostra como passar configurações para a sessão do Spark.

Você pode usar a opção --config para especificar vários parâmetros de configuração. Você também pode especificá-las na sessão do aplicativo definindo a configuração no objeto 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

O comando curl executa o aplicativo usando Livy. Substitua USER, PASSWORD e LIVY_ENDPOINT para refletir seu ambiente.

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

O comando azdata executa o aplicativo usando parâmetros comumente especificados. Para obter opções de parâmetro completas para azdata bdc spark batch create, consulte azdata bdc spark.

O aplicativo requer o parâmetro de configuração spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation. Portanto, o comando usa a opção --config. Essa configuração mostra como passar configurações para a sessão do Spark.

Você pode usar a opção --config para especificar vários parâmetros de configuração. Você também pode especificá-las na sessão do aplicativo definindo a configuração no objeto 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

O comando curl executa o aplicativo usando Livy. Substitua USER, PASSWORD e LIVY_ENDPOINT para refletir seu ambiente.

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

O comando azdata executa o aplicativo usando parâmetros comumente especificados. Para obter opções de parâmetro completas para azdata bdc spark batch create, consulte azdata bdc spark.

Como o exemplo do PySpark, esse aplicativo também requer o parâmetro de configuração spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation. Portanto, o comando usa a opção --config. Essa configuração mostra como passar configurações para a sessão do Spark.

Você pode usar a opção --config para especificar vários parâmetros de configuração. Você também pode especificá-las na sessão do aplicativo definindo a configuração no objeto 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

O comando curl executa o aplicativo usando Livy. Substitua USER, PASSWORD e LIVY_ENDPOINT para refletir seu ambiente.

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