Übermitteln von Spark-Aufträgen mithilfe von Befehlszeilentools

Gilt für: SQL Server 2019 (15.x)

Dieser Artikel enthält Anleitungen zur Verwendung von Befehlszeilentools zum Ausführen von Spark-Aufträgen auf SQL Server-Big Data-Cluster.

Wichtig

Das Microsoft SQL Server 2019-Big Data-Cluster-Add-On wird eingestellt. Der Support für SQL Server 2019-Big Data-Clusters endet am 28. Februar 2025. Alle vorhandenen Benutzer*innen von SQL Server 2019 mit Software Assurance werden auf der Plattform vollständig unterstützt, und die Software wird bis zu diesem Zeitpunkt weiterhin über kumulative SQL Server-Updates verwaltet. Weitere Informationen finden Sie im Ankündigungsblogbeitrag und unter Big Data-Optionen auf der Microsoft SQL Server-Plattform.

Voraussetzungen

• SQL Server 2019 Big Data-Tools konfiguriert und beim Cluster angemeldet:
  • azdata
  • Eine curl-Anwendung zum Ausführen von REST-API-Aufrufen an Livy

Spark-Aufträge, die azdata oder Livy verwenden

Dieser Artikel enthält Beispiele für die Verwendung von Befehlszeilenmustern zum Übermitteln von Spark-Anwendungen an SQL Server-Big Data-Cluster.

Mit den azdata bdc sparkBefehlen der Azure Data-CLI können alle Funktionen von Spark-SQL Server-Big Data-Cluster in der Befehlszeile ausgeführt werden. Dieser Artikel konzentriert sich auf die Auftragsübermittlung. Aber auch azdata bdc spark unterstützt über den Befehl azdata bdc spark sessioninteraktive Modi für Python, Scala, SQL und R.

Wenn Sie direkte Integration in eine REST-API benötigen, verwenden Sie standardmäßige Livy-Aufrufe zum Übermitteln von Aufträgen. In diesem Artikel wird das Befehlszeilentool curlin den Livy-Beispielen verwendet, um den REST-API-Aufruf auszuführen. Ein detailliertes Beispiel für die Interaktion mit dem Spark Livy-Endpunkt mithilfe von Python-Code finden Sie auf GitHub unter Using Spark from the Livy endpoint (Verwenden von Spark vom Livy-Endpunkt aus).

Einfaches ETL, bei dem Spark für Big Data-Cluster verwendet wird

Diese ETL-Anwendung (Extrahieren, Transformieren und Laden) folgt einem häufigen Datentechnikmuster. Sie lädt Tabellendaten aus einem Zielzonenpfad von Apache Hadoop Distributed File System (HDFS). Anschließend verwendet sie ein Tabellenformat, um in einen von HDFS verarbeiteten Zonenpfad zu schreiben.

Laden Sie das Dataset für die Beispielanwendung herunter. Erstellen Sie dann PySpark-Anwendungen mithilfe von PySpark, Spark Scala oder Spark SQL.

In den folgenden Abschnitten finden Sie Beispielübungen für die einzelnen Lösungen. Wählen Sie die Registerkarte für Ihre Plattform aus. Sie werden die Anwendung mithilfe von azdata oder curl ausführen.

PySpark

In diesem Beispiel wird die folgende PySpark-Anwendung verwendet. Sie wird auf dem lokalen Computer als die Python-Datei parquet_etl_sample.py gespeichert.

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

Kopieren der PySpark-Anwendung in HDFS

Speichern Sie die Anwendung in HDFS, damit der Cluster zur Ausführung darauf zugreifen kann. Als bewährte Methode sollten Sie Anwendungsspeicherorte innerhalb des Clusters standardisieren und steuern, um die Verwaltung zu vereinfachen.

In diesem Beispiel für einen Anwendungsfall werden alle ETL-Pipelineanwendungen im Pfad hdfs:/apps/ETL-Pipelines gespeichert. Die Beispielanwendung wird unter hdfs:/apps/ETL-Pipelines/parquet_etl_sample.py gespeichert.

Führen Sie den folgenden Befehl aus, um parquet_etl_sample.py vom lokalen Entwicklungs- oder Stagingcomputer in den HDFS-Cluster hochzuladen.

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

Spark Scala

In diesem Beispiel wird eine Spark-Anwendung verwendet, die in Scala Spark geschrieben wurde.

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()
    }
}

Bündeln der Spark-Anwendung und Kopieren in HDFS

In der Spark-Dokumentation wird die Erstellung einer Assembly-JAR-Datei (oder eines Bündels) empfohlen, in der/dem Ihre Anwendung und sämtliche Abhängigkeiten enthalten sind. Dieser Schritt ist erforderlich, um das Anwendungsbündel zur Ausführung an den Cluster zu übermitteln.

Das Einrichten einer vollständigen Scala Spark-Entwicklungsumgebung geht über den Rahmen dieses Artikels hinaus. Weitere Informationen finden Sie in der Spark-Dokumentation zum Erstellen eigenständiger Anwendungen.

In diesem Beispiel wird davon ausgegangen, dass das JAR-Bündel parquet-etl-sample.jar der Anwendung kompiliert wurde und zur Verfügung steht. Führen Sie den folgenden Befehl aus, um das Paket vom lokalen Entwicklungs- oder Stagingcomputer in den HDFS-Cluster hochzuladen.

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

Spark SQL

In diesem Beispiel wird Spark SQL für die Erfassungslogik verwendet. Es werden darin Tabellen und Sichten verwendet, um einen SQL-zentrierten Ansatz für ETL zu bieten.

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;

Kopieren der Spark-SQL-Anwendung in HDFS

Führen Sie den folgenden Befehl aus, um die Datei parquet_etl_sample.sql vom lokalen Entwicklungs- oder Stagingcomputer in den HDFS-Cluster hochzuladen.

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

Ausführen der Spark-Anwendung

Verwenden Sie den folgenden Befehl, um die Anwendung zur Ausführung an Spark-SQL Server-Big Data-Cluster zu übermitteln.

PySpark und azdata

Der Befehl azdata führt die Anwendung mithilfe von häufig angegebenen Parametern aus. Umfassende Parameteroptionen für azdata bdc spark batch create finden Sie unter azdata bdc spark.

Für diese Anwendung ist der Konfigurationsparameter spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation erforderlich. Deshalb wird im Befehl die Option --config verwendet. Dieses Setup zeigt, wie Konfigurationen in die Spark-Sitzung übergeben werden.

Sie können mithilfe der Option --config mehrere Konfigurationsparameter angeben. Sie könnten sie auch im Rahmen der Anwendungssitzung angeben, indem Sie die Konfiguration im Objekt SparkSession festlegen.

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

Warnung

Der Parameter „name“ oder „n“ für den Batchnamen muss bei jeder Erstellung eines neuen Batches eindeutig sein.

PySpark und curl, mit Livy

Der Befehl curl führt die Anwendung mithilfe von Livy aus. Ersetzen Sie USER, PASSWORD und LIVY_ENDPOINT, um Ihre Umgebung widerzuspiegeln.

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

Warnung

Der Parameter „name“ muss bei jeder Erstellung eines neuen Batches eindeutig sein.

Scala und azdata

Der Befehl azdata führt die Anwendung mithilfe von häufig angegebenen Parametern aus. Umfassende Parameteroptionen für azdata bdc spark batch create finden Sie unter azdata bdc spark.

Für die Anwendung ist der Konfigurationsparameter spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation erforderlich. Deshalb wird im Befehl die Option --config verwendet. Dieses Setup zeigt, wie Konfigurationen in die Spark-Sitzung übergeben werden.

Sie können mithilfe der Option --config mehrere Konfigurationsparameter angeben. Sie könnten sie auch im Rahmen der Anwendungssitzung angeben, indem Sie die Konfiguration im Objekt SparkSession festlegen.

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

Warnung

Der Parameter „name“ oder „n“ für den Batchnamen muss bei jeder Erstellung eines neuen Batches eindeutig sein.

Scala und curl, mit Livy

Der Befehl curl führt die Anwendung mithilfe von Livy aus. Ersetzen Sie USER, PASSWORD und LIVY_ENDPOINT, um Ihre Umgebung widerzuspiegeln.

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

Warnung

Der Parameter „name“ für den Batchnamen muss bei jeder Erstellung eines neuen Batches eindeutig sein.

SQL und azdata

Der Befehl azdata führt die Anwendung mithilfe von häufig angegebenen Parametern aus. Umfassende Parameteroptionen für azdata bdc spark batch create finden Sie unter azdata bdc spark.

Wie das PySpark-Beispiel erfordert auch diese Anwendung den Konfigurationsparameter spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation. Deshalb wird im Befehl die Option --config verwendet. Dieses Setup zeigt, wie Konfigurationen in die Spark-Sitzung übergeben werden.

Sie können mithilfe der Option --config mehrere Konfigurationsparameter angeben. Sie könnten sie auch im Rahmen der Anwendungssitzung angeben, indem Sie die Konfiguration im Objekt SparkSession festlegen.

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

Warnung

Der Parameter „name“ oder „n“ für den Batchnamen muss bei jeder Erstellung eines neuen Batches eindeutig sein.

SQL und curl, mit Livy

Der Befehl curl führt die Anwendung mithilfe von Livy aus. Ersetzen Sie USER, PASSWORD und LIVY_ENDPOINT, um Ihre Umgebung widerzuspiegeln.

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

Warnung

Der Parameter „name“ muss bei jeder Erstellung eines neuen Batches eindeutig sein.

Überwachen von Spark-Aufträgen

Die azdata bdc spark batch-Befehle bieten Verwaltungsaktionen für Spark-Batchaufträge.

Führen Sie den folgenden Befehl aus, um alle gerade ausgeführten Aufträge aufzulisten.

• Mit dem Befehl azdata:

  azdata bdc spark batch list -o table
  

• Der Befehl curl, mit Livy:

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

Führen Sie zum Abrufen von Informationen für einen Spark-Batch mit der angegebenen ID den folgenden Befehl aus. Die batch id wird aus spark batch create zurückgegeben.

• Mit dem Befehl azdata:

  azdata bdc spark batch info --batch-id 0
  

• Der Befehl curl, mit Livy:

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

Führen Sie zum Abrufen von Statusinformationen für einen Spark-Batch mit der angegebenen ID den folgenden Befehl aus.

• Mit dem Befehl azdata:

  azdata bdc spark batch state --batch-id 0
  

• Der Befehl curl, mit Livy:

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

Führen Sie zum Abrufen der Protokolle für einen Spark-Batch mit der angegebenen ID den folgenden Befehl aus.

• Mit dem Befehl azdata:

  azdata bdc spark batch log --batch-id 0
  

• Der Befehl curl, mit Livy:

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

Nächste Schritte

Informationen zur Problembehandlung bei Spark-Code finden Sie unter Problembehandlung für ein PySpark-Notebook.

Umfassenden Spark-Beispielcode finden Sie auf GitHub unter SQL Server Big Data Clusters Spark samples (Beispiele für Spark-SQL Server-Big Data-Cluster).

Weitere Informationen zu SQL Server-Big Data-Cluster und zugehörige Szenarios finden Sie unter Big Data-Cluster für SQL Server.