コマンド ライン ツールを使用して Spark ジョブを送信する

この例では、次の PySpark アプリケーションを使用します。 ローカル コンピューターに parquet_etl_sample.py という名前の Python ファイルとして保存されます。

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

PySpark アプリケーションを HDFS にコピーする

クラスターが実行のためにアプリケーションにアクセスできるように、アプリケーションを HDFS に格納します。 ベスト プラクティスとして、クラスター内のアプリケーションの場所を標準化して管理し、管理を合理化します。

このユース ケースの例では、すべての ETL パイプライン アプリケーションが hdfs:/apps/ETL-Pipelines パスに格納されます。 サンプル アプリケーションは hdfs:/apps/ETL-Pipelines/parquet_etl_sample.py に格納されます。

次のコマンドを実行して 、 ローカル開発またはステージング コンピューターから HDFS クラスターにparquet_etl_sample.pyをアップロードします。

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

この例では、Scala Spark で記述された 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()
    }
}

Spark アプリケーションをバンドルして HDFS にコピーする

Spark ドキュメントでは、アプリケーションとすべての依存関係を含む アセンブリ JAR (またはバンドル) を作成することをお勧めします。 この手順は、アプリケーション バンドルをクラスターに送信して実行するために必要です。

完全な Scala Spark 開発環境の設定は、この記事の範囲外です。 詳細については、 自己完結型アプリケーションを作成するための Spark ドキュメントを参照してください。

この例では、parquet-etl-sample.jarという名前 の アプリケーション JAR バンドルがコンパイルされ、使用可能であることを前提としています。 次のコマンドを実行して、ローカル開発またはステージング マシンから HDFS クラスターにバンドルをアップロードします。

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

この例では、取り込みロジックに Spark SQL を使用します。 テーブルとビューを使用して、ETL に対する SQL 中心のアプローチを提供します。

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;

Spark SQL アプリケーションを HDFS にコピーする

次のコマンドを実行して、 parquet-etl-sample.sql ファイルをローカル開発またはステージング コンピューターから HDFS クラスターにアップロードします。

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

azdata コマンドは、一般的に指定されたパラメーターを使用してアプリケーションを実行します。 azdata bdc spark batch createの完全なパラメーター オプションについては、azdata bdc sparkを参照してください。

このアプリケーションには、 spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation 構成パラメーターが必要です。 そのため、コマンドは --config オプションを使用します。 このセットアップでは、Spark セッションに構成を渡す方法を示します。

--config オプションを使用して、複数の構成パラメーターを指定できます。 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

curl コマンドは、Livy を使用してアプリケーションを実行します。 環境を反映するために、 USER、 PASSWORD、および LIVY_ENDPOINT を置き換えます。

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

azdata コマンドは、一般的に指定されたパラメーターを使用してアプリケーションを実行します。 azdata bdc spark batch createの完全なパラメーター オプションについては、azdata bdc sparkを参照してください。

アプリケーションには、 spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation 構成パラメーターが必要です。 そのため、コマンドは --config オプションを使用します。 このセットアップでは、Spark セッションに構成を渡す方法を示します。

--config オプションを使用して、複数の構成パラメーターを指定できます。 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

curl コマンドは、Livy を使用してアプリケーションを実行します。 環境を反映するために、 USER、 PASSWORD、および LIVY_ENDPOINT を置き換えます。

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

azdata コマンドは、一般的に指定されたパラメーターを使用してアプリケーションを実行します。 azdata bdc spark batch createの完全なパラメーター オプションについては、azdata bdc sparkを参照してください。

PySpark の例と同様に、このアプリケーションにも spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation 構成パラメーターが必要です。 そのため、コマンドは --config オプションを使用します。 このセットアップでは、Spark セッションに構成を渡す方法を示します。

--config オプションを使用して、複数の構成パラメーターを指定できます。 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

curl コマンドは、Livy を使用してアプリケーションを実行します。 環境を反映するために、 USER、 PASSWORD、および LIVY_ENDPOINT を置き換えます。

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