Senden von Spark-Aufträgen mithilfe von Befehlszeilentools

In diesem Beispiel wird die folgende PySpark-Anwendung verwendet. Sie wird als Python-Datei mit dem Namen parquet_etl_sample.py auf dem lokalen Computer 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 können Sie Anwendungsstandorte innerhalb des Clusters standardisieren und steuern, um die Verwaltung zu optimieren.

In diesem Beispielfall werden alle ETL-Pipelineanwendungen auf dem 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"

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 und Kopieren der Spark-Anwendung in HDFS

Die Spark-Dokumentation empfiehlt das Erstellen einer Assembly JAR (oder eines Bündels), die Ihre Anwendung und alle Abhängigkeiten enthält. Dieser Schritt ist erforderlich, um das Anwendungsbundle 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 ein Anwendungs-JAR-Bündel mit dem Namen parquet-etl-sample.jar kompiliert und verfügbar ist. Führen Sie den folgenden Befehl aus, um das Bundle 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"

In diesem Beispiel wird Spark SQL für die Aufnahmelogik verwendet. Es verwendet Tabellen und Ansichten, um einen SQL-zentrierten Ansatz für ETL bereitzustellen.

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 parquet-etl-sample.sql Datei 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"

Der azdata Befehl führt die Anwendung mithilfe häufig angegebener Parameter aus. Vollständige Parameteroptionen für azdata bdc spark batch create, siehe azdata bdc spark.

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

Sie können die --config Option verwenden, um mehrere Konfigurationsparameter anzugeben. Sie können sie auch innerhalb der Anwendungssitzung angeben, indem Sie die Konfiguration im SparkSession Objekt 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

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

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

Der azdata Befehl führt die Anwendung mithilfe häufig angegebener Parameter aus. Vollständige Parameteroptionen für azdata bdc spark batch create, siehe azdata bdc spark.

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

Sie können die --config Option verwenden, um mehrere Konfigurationsparameter anzugeben. Sie können sie auch innerhalb der Anwendungssitzung angeben, indem Sie die Konfiguration im SparkSession Objekt 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

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

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

Der azdata Befehl führt die Anwendung mithilfe häufig angegebener Parameter aus. Vollständige Parameteroptionen für azdata bdc spark batch create, siehe azdata bdc spark.

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

Sie können die --config Option verwenden, um mehrere Konfigurationsparameter anzugeben. Sie können sie auch innerhalb der Anwendungssitzung angeben, indem Sie die Konfiguration im SparkSession Objekt 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

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

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