Enviar trabalhos do Spark usando ferramentas de linha de comando

Aplica-se a: SQL Server 2019 (15.x)

Este artigo fornece diretrizes sobre como usar as ferramentas de linha de comando para executar trabalhos do Spark nos Clusters de Big Data do SQL Server.

Importante

O complemento Clusters de Big Data do Microsoft SQL Server 2019 será desativado. O suporte para Clusters de Big Data do SQL Server 2019 será encerrado em 28 de fevereiro de 2025. Todos os usuários existentes do SQL Server 2019 com Software Assurance terão suporte total na plataforma e o software continuará a ser mantido por meio de atualizações cumulativas do SQL Server até esse momento. Para obter mais informações, confira a postagem no blog de anúncio e as opções de Big Data na plataforma do Microsoft SQL Server.

Pré-requisitos

• Ferramentas de Big Data do SQL Server 2019 configuradas e conectadas ao cluster:
  • azdata
  • Um aplicativo curl para executar chamadas à API REST para Livy

Trabalhos do Spark que usam azdata ou Livy

Este artigo fornece exemplos de como usar padrões de linha de comando para enviar aplicativos Spark para Clusters de Big Data do SQL Server.

Os comandos azdata bdc spark da CLI de dados do Azure abrangem todas as funcionalidades do Spark nos Clusters de Big Data do SQL Server na linha de comando. Este artigo se concentra no envio de trabalhos. No entanto, o azdata bdc spark também dá suporte a modos interativos para Python, Scala, SQL e R por meio do comando azdata bdc spark session.

Se você precisa de integração direta com uma API REST, use chamadas Livy padrão para enviar trabalhos. Este artigo usa a ferramenta de linha de comando curl nos exemplos do Livy para executar a chamada à API REST. Para obter um exemplo detalhado que mostra como interagir com o ponto de extremidade do Livy no Spark usando o código do Python, confira Usar o Spark do ponto de extremidade do Livy no GitHub.

ETL simples que usa o Spark dos Clusters de Big Data

Esse aplicativo ETL (extração, transformação e carregamento) segue um padrão de engenharia de dados comum. Ele carrega dados tabulares de um caminho de zona de destino de HDFS (Sistema de Arquivos Distribuído do Apache Hadoop). Em seguida, usa um formato de tabela para gravar em um caminho de zona processado pelo HDFS.

Baixe o conjunto de dados do aplicativo de exemplo. Em seguida, crie aplicativos PySpark usando PySpark, Spark Scala ou Spark SQL.

Nas seções a seguir, você encontrará exercícios de exemplo para cada solução. Selecione a guia para sua plataforma. Você executará o aplicativo usando azdata ou curl.

PySpark

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"

Spark Scala

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"

Spark SQL

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"

Executar o aplicativo Spark

Use o comando a seguir para enviar o aplicativo para o Spark dos Clusters de Big Data do SQL Server para execução.

PySpark e azdata

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

Aviso

O parâmetro "name" ou "n" para o nome do lote deverá ser exclusivo cada vez que um novo lote for criado.

PySpark e curl, usando Livy

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

Aviso

O parâmetro "name" deverá ser exclusivo cada vez que um novo lote for criado.

Scala e azdata

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

Aviso

O parâmetro "name" ou "n" para o nome do lote deverá ser exclusivo cada vez que um novo lote for criado.

Scala e curl, usando Livy

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

Aviso

O parâmetro "name" para o nome do lote deverá ser exclusivo cada vez que um novo lote for criado.

SQL e azdata

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

Aviso

O parâmetro "name" ou "n" para o nome do lote deverá ser exclusivo cada vez que um novo lote for criado.

SQL e curl, usando Livy

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

Aviso

O parâmetro "name" deverá ser exclusivo cada vez que um novo lote for criado.

Monitorar trabalhos do Spark

Os comandos azdata bdc spark batch contêm ações de gerenciamento para trabalhos em lote do Spark.

Para listar todos os trabalhos em execução, aplique o comando a seguir.

• O comando azdata:

  azdata bdc spark batch list -o table
  

• O comando curl, usando Livy:

  curl -k -u <USER>:<PASSWORD> -X POST <LIVY_ENDPOINT>/batches
  

Para obter informações para um lote do Spark com a ID fornecida, execute o comando a seguir. O batch id é retornado de spark batch create.

• O comando azdata:

  azdata bdc spark batch info --batch-id 0
  

• O comando curl, usando Livy:

  curl -k -u <USER>:<PASSWORD> -X POST <LIVY_ENDPOINT>/batches/<BATCH_ID>
  

Para obter informações de estado de um lote do Spark com a ID fornecida, execute o comando a seguir.

• O comando azdata:

  azdata bdc spark batch state --batch-id 0
  

• O comando curl, usando Livy:

  curl -k -u <USER>:<PASSWORD> -X POST <LIVY_ENDPOINT>/batches/<BATCH_ID>/state
  

Para obter os logs de um lote do Spark com a ID fornecida, execute o comando a seguir.

• O comando azdata:

  azdata bdc spark batch log --batch-id 0
  

• O comando curl, usando Livy:

  curl -k -u <USER>:<PASSWORD> -X POST <LIVY_ENDPOINT>/batches/<BATCH_ID>/log
  

Próximas etapas

Para obter mais informações sobre como solucionar problemas de código do Spark, confira Solucionar problemas de um notebook PySpark.

Um abrangente código de exemplo do Spark está disponível em Exemplos do Spark de Clusters de Big Data do SQL Server no GitHub.

Para obter mais informações sobre os Clusters de Big Data do SQL Server e os cenários relacionados, confira Clusters de Big Data do SQL Server.