Mengirimkan pekerjaan Spark dengan menggunakan alat baris perintah

Berlaku untuk: SQL Server 2019 (15.x)

Artikel ini menyediakan panduan tentang cara menggunakan alat baris perintah untuk menjalankan pekerjaan Spark di SQL Server Kluster Big Data.

Penting

Add-on Kluster Big Data Microsoft SQL Server 2019 akan dihentikan. Dukungan untuk SQL Server 2019 Kluster Big Data akan berakhir pada 28 Februari 2025. Semua pengguna SQL Server 2019 yang ada dengan Jaminan Perangkat Lunak akan didukung sepenuhnya pada platform dan perangkat lunak akan terus dipertahankan melalui pembaruan kumulatif SQL Server hingga saat itu. Untuk informasi selengkapnya, lihat posting blog pengumuman dan Opsi big data di platform Microsoft SQL Server.

Prasyarat

• Alat big data SQL Server 2019 dikonfigurasi dan masuk ke kluster:
  • azdata
  • Aplikasi curl untuk melakukan panggilan REST API ke Livy

Pekerjaan Spark yang menggunakan azdata atau Livy

Artikel ini menyediakan contoh cara menggunakan pola baris perintah untuk mengirimkan aplikasi Spark ke SQL Server Kluster Big Data.

Perintah Azure Data CLI azdata bdc spark menampilkan semua kemampuan SQL Server Kluster Big Data Spark pada baris perintah. Artikel ini berfokus pada pengiriman pekerjaan. Tetapi azdata bdc spark juga mendukung mode interaktif untuk Python, Scala, SQL, dan R melalui azdata bdc spark session perintah .

Jika Anda memerlukan integrasi langsung dengan REST API, gunakan panggilan Livy standar untuk mengirimkan pekerjaan. Artikel ini menggunakan curl alat baris perintah dalam contoh Livy untuk menjalankan panggilan REST API. Untuk contoh terperinci yang menunjukkan cara berinteraksi dengan titik akhir Spark Livy dengan menggunakan kode Python, lihat Menggunakan Spark dari titik akhir Livy di GitHub.

ETL sederhana yang menggunakan Kluster Big Data Spark

Aplikasi ekstrak, transformasi, dan pemuatan (ETL) ini mengikuti pola rekayasa data umum. Ini memuat data tabular dari jalur zona landing Apache Hadoop Distributed File System (HDFS). Kemudian menggunakan format tabel untuk menulis ke jalur zona yang diproses HDFS.

Unduh himpunan data aplikasi sampel. Kemudian buat aplikasi PySpark dengan menggunakan PySpark, Spark Scala, atau Spark SQL.

Di bagian berikut, Anda akan menemukan sampel latihan untuk setiap solusi. Pilih tab untuk platform Anda. Anda akan menjalankan aplikasi dengan menggunakan azdata atau curl.

PySpark

Contoh ini menggunakan aplikasi PySpark berikut. Ini disimpan sebagai file Python bernama parquet_etl_sample.py di komputer lokal.

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

Menyalin aplikasi PySpark ke HDFS

Simpan aplikasi dalam HDFS sehingga kluster dapat mengaksesnya untuk eksekusi. Sebagai praktik terbaik, standarisasi dan tata lokasi aplikasi dalam kluster untuk menyederhanakan administrasi.

Dalam contoh kasus penggunaan ini, semua aplikasi alur ETL disimpan di jalur hdfs:/apps/ETL-Pipelines . Aplikasi sampel disimpan di hdfs:/apps/ETL-Pipelines/parquet_etl_sample.py.

Jalankan perintah berikut untuk mengunggah parquet_etl_sample.py dari pengembangan lokal atau mesin penahapan ke kluster HDFS.

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

Spark Scala

Contoh ini menggunakan aplikasi Spark yang ditulis dalam 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()
    }
}

Bundel dan salin aplikasi Spark ke HDFS

Dokumentasi Spark merekomendasikan pembuatan JAR rakitan (atau bundel) yang berisi aplikasi Anda dan semua dependensi. Langkah ini diperlukan untuk mengirimkan bundel aplikasi ke kluster untuk eksekusi.

Menyiapkan lingkungan pengembangan Scala Spark lengkap berada di luar cakupan artikel ini. Untuk informasi selengkapnya, lihat dokumentasi Spark untuk membuat aplikasi mandiri.

Contoh ini mengasumsikan bahwa bundel JAR aplikasi bernama parquet-etl-sample.jar dikompilasi dan tersedia. Jalankan perintah berikut untuk mengunggah bundel dari pengembangan lokal atau mesin penahapan ke kluster HDFS.

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

Spark SQL

Contoh ini menggunakan Spark SQL untuk logika penyerapan. Ini menggunakan tabel dan tampilan untuk menyediakan pendekatan yang ber sentris SQL untuk 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;

Menyalin aplikasi Spark SQL ke HDFS

Jalankan perintah berikut untuk mengunggah file parquet-etl-sample.sql dari komputer pengembangan atau penahapan lokal ke kluster HDFS.

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

Menjalankan aplikasi Spark

Gunakan perintah berikut untuk mengirimkan aplikasi ke SQL Server Kluster Big Data Spark untuk eksekusi.

PySpark dan azdata

Perintah azdata menjalankan aplikasi dengan menggunakan parameter yang umum ditentukan. Untuk opsi parameter lengkap untuk azdata bdc spark batch create, lihat azdata bdc spark.

Aplikasi ini memerlukan spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation parameter konfigurasi. Jadi perintah menggunakan --config opsi . Penyiapan ini menunjukkan cara meneruskan konfigurasi ke sesi Spark.

Anda dapat menggunakan --config opsi untuk menentukan beberapa parameter konfigurasi. Anda juga dapat menentukannya di dalam sesi aplikasi dengan mengatur konfigurasi dalam SparkSession objek.

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

Peringatan

Parameter "name" atau "n" untuk nama batch harus unik setiap kali batch baru dibuat.

PySpark dan curl, menggunakan Livy

Perintah curl menjalankan aplikasi dengan menggunakan Livy. Ganti , PASSWORD, dan LIVY_ENDPOINT untuk mencerminkan USERlingkungan Anda.

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

Peringatan

Parameter "nama" harus unik setiap kali batch baru dibuat.

Scala dan azdata

Perintah azdata menjalankan aplikasi dengan menggunakan parameter yang umum ditentukan. Untuk opsi parameter lengkap untuk azdata bdc spark batch create, lihat azdata bdc spark.

Aplikasi ini memerlukan spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation parameter konfigurasi. Jadi perintah menggunakan --config opsi . Penyiapan ini menunjukkan cara meneruskan konfigurasi ke sesi Spark.

Anda dapat menggunakan --config opsi untuk menentukan beberapa parameter konfigurasi. Anda juga dapat menentukannya di dalam sesi aplikasi dengan mengatur konfigurasi dalam SparkSession objek.

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

Peringatan

Parameter "name" atau "n" untuk nama batch harus unik setiap kali batch baru dibuat.

Scala dan curl, menggunakan Livy

Perintah curl menjalankan aplikasi dengan menggunakan Livy. Ganti , PASSWORD, dan LIVY_ENDPOINT untuk mencerminkan USERlingkungan Anda.

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

Peringatan

Parameter "nama" untuk nama batch harus unik setiap kali batch baru dibuat.

SQL dan azdata

Perintah azdata menjalankan aplikasi dengan menggunakan parameter yang umum ditentukan. Untuk opsi parameter lengkap untuk azdata bdc spark batch create, lihat azdata bdc spark.

Seperti contoh PySpark, aplikasi ini juga memerlukan spark.sql.legacy.allowCreatingManagedTableUsingNonemptyLocation parameter konfigurasi. Jadi perintah menggunakan --config opsi . Penyiapan ini menunjukkan cara meneruskan konfigurasi ke sesi Spark.

Anda dapat menggunakan --config opsi untuk menentukan beberapa parameter konfigurasi. Anda juga dapat menentukannya di dalam sesi aplikasi dengan mengatur konfigurasi dalam SparkSession objek.

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

Peringatan

Parameter "name" atau "n" untuk nama batch harus unik setiap kali batch baru dibuat.

SQL dan curl, menggunakan Livy

Perintah curl menjalankan aplikasi dengan menggunakan Livy. Ganti , PASSWORD, dan LIVY_ENDPOINT untuk mencerminkan USERlingkungan Anda.

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

Peringatan

Parameter "nama" harus unik setiap kali batch baru dibuat.

Memantau pekerjaan Spark

Perintah menyediakan azdata bdc spark batch tindakan manajemen untuk pekerjaan batch Spark.

Untuk mencantumkan semua pekerjaan yang sedang berjalan, jalankan perintah berikut.

• Perintah azdata:

  azdata bdc spark batch list -o table
  

• curl Perintah , menggunakan Livy:

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

Untuk mendapatkan informasi untuk batch Spark dengan ID yang diberikan, jalankan perintah berikut. batch id dikembalikan dari spark batch create.

• Perintah azdata:

  azdata bdc spark batch info --batch-id 0
  

• curl Perintah , menggunakan Livy:

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

Untuk mendapatkan informasi status untuk batch Spark dengan ID yang diberikan, jalankan perintah berikut.

• Perintah azdata:

  azdata bdc spark batch state --batch-id 0
  

• curl Perintah , menggunakan Livy:

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

Untuk mendapatkan log untuk batch Spark dengan ID yang diberikan, jalankan perintah berikut.

• Perintah azdata:

  azdata bdc spark batch log --batch-id 0
  

• curl Perintah , menggunakan Livy:

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

Langkah berikutnya

Untuk informasi tentang pemecahan masalah kode Spark, lihat Memecahkan masalah buku catatan PySpark.

Kode sampel Spark komprehensif tersedia di SQL Server Kluster Big Data sampel Spark di GitHub.

Untuk informasi selengkapnya tentang Kluster Big Data SQL Server dan skenario terkait, lihat SQL Server Kluster Big Data.