Enviar trabalhos do Spark usando ferramentas de linha de comando

Este exemplo usa o aplicativo 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 o HDFS

Armazene o aplicativo no HDFS para que o cluster possa acessá-lo para execução. Como prática recomendada, padronizar e controlar os locais do aplicativo dentro do cluster para simplificar a administração.

Neste caso de uso de exemplo, todos os aplicativos de pipeline ETL sã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 parquet_etl_sample.py do computador de preparo ou desenvolvimento local para o 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 para o HDFS

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

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

Este exemplo pressupõe que um pacote JAR de aplicativo chamado parquet-etl-sample.jar esteja compilado e disponível. Execute o comando a seguir para carregar o pacote do computador de preparo ou desenvolvimento local para o 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 SQL do Spark 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 SQL do Spark para o HDFS

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

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

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

Este aplicativo requer o spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation parâmetro de configuração. 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á-los dentro da sessão do aplicativo definindo a configuração no SparkSession objeto.

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 curl comando executa o aplicativo usando Livy. Substitua USERe PASSWORDLIVY_ENDPOINT reflita 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 azdata comando executa o aplicativo usando parâmetros normalmente especificados. Para obter opções de parâmetro completas para azdata bdc spark batch create, consulte azdata bdc spark.

O aplicativo requer o spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation parâmetro de configuração. 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á-los dentro da sessão do aplicativo definindo a configuração no SparkSession objeto.

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 curl comando executa o aplicativo usando Livy. Substitua USERe PASSWORDLIVY_ENDPOINT reflita 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 azdata comando executa o aplicativo usando parâmetros normalmente especificados. Para obter opções de parâmetro completas para azdata bdc spark batch create, consulte azdata bdc spark.

Assim como o exemplo do PySpark, esse aplicativo também requer o spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation parâmetro de configuração. 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á-los dentro da sessão do aplicativo definindo a configuração no SparkSession objeto.

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 curl comando executa o aplicativo usando Livy. Substitua USERe PASSWORDLIVY_ENDPOINT reflita 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