명령줄 도구를 사용하여 Spark 작업 제출

적용 대상: SQL Server 2019(15.x)

이 문서에서는 명령줄 도구를 사용하여 SQL Server 빅 데이터 클러스터에서 Spark 작업을 실행하는 방법에 대한 지침을 제공합니다.

중요

Microsoft SQL Server 2019 빅 데이터 클러스터 추가 기능이 사용 중지됩니다. SQL Server 2019 빅 데이터 클러스터에 대한 지원은 2025년 2월 28일에 종료됩니다. Software Assurance를 사용하는 SQL Server 2019의 모든 기존 사용자는 플랫폼에서 완전히 지원되며, 소프트웨어는 해당 시간까지 SQL Server 누적 업데이트를 통해 계속 유지 관리됩니다. 자세한 내용은 공지 블로그 게시물 및 Microsoft SQL Server 플랫폼의 빅 데이터 옵션을 참조하세요.

사전 요구 사항

• 다음 SQL Server 2019 빅 데이터 도구를 구성하고 클러스터에 로그인합니다.
  • azdata
  • Livy에 대한 REST API 호출을 수행하기 위한 curl 애플리케이션

azdata 또는 Livy를 사용하는 Spark 작업

이 문서에는 명령줄 패턴을 사용하여 SQL Server 빅 데이터 클러스터에 Spark 애플리케이션을 제출하는 방법의 예제가 나와 있습니다.

Azure Data CLI azdata bdc spark 명령은 명령줄에 SQL Server 빅 데이터 클러스터 Spark의 모든 기능을 표시합니다. 이 문서에서는 주로 작업 제출에 대해 다룹니다. 하지만 azdata bdc spark는 azdata bdc spark session 명령을 통해 Python, Scala, SQL, R에 대한 대화형 모드도 지원합니다.

REST API에 직접 통합해야 할 경우 표준 Livy 호출을 사용하여 작업을 제출하세요. 이 문서에서는 Livy 예제에서 curl 명령줄 도구를 사용하여 REST API 호출을 실행합니다. Python 코드를 사용하여 Spark Livy 엔드포인트와 상호 작용하는 방법을 보여주는 자세한 예제는 GitHub의 Use Spark from the Livy endpoint(Livy 엔드포인트에서 Spark 사용)를 참조하세요.

빅 데이터 클러스터 Spark를 사용하는 단순 ETL

ETL(추출, 변환 및 로드) 애플리케이션은 일반적인 데이터 엔지니어링 패턴을 따릅니다. HDFS(Apache Hadoop 분산 파일 시스템) 랜딩 존 경로에서 표 형식 데이터를 로드합니다. 그런 다음 테이블 형식을 사용하여 HDFS로 처리된 영역 경로에 씁니다.

샘플 애플리케이션의 데이터 세트를 다운로드합니다. 그런 다음 PySpark, Spark Scala 또는 Spark SQL을 사용하여 PySpark 애플리케이션을 만듭니다.

다음 섹션에는 각 솔루션에 대한 연습 문제 샘플을 찾을 수 있습니다. 플랫폼에 대한 탭을 선택합니다. azdata 또는 curl을 사용하여 애플리케이션을 실행하게 됩니다.

PySpark

이 예제에서는 다음 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"

Spark Scala

이 예제에서는 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

이 예제에서는 수집 논리에 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에 복사

다음 명령을 실행하여 로컬 개발 또는 준비 머신에서 HDFS 클러스터로 parquet-etl-sample.sql 파일을 업로드합니다.

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

Spark 애플리케이션 실행

다음 명령을 사용하여 실행을 위해 애플리케이션을 SQL Server 빅 데이터 클러스터 Spark에 제출합니다.

PySpark 및 azdata

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

경고

일괄 처리 이름의 "name" 또는 "n" 매개 변수는 새 일괄 처리를 만들 때마다 고유해야 합니다.

Livy를 사용하는 PySpark 및 curl

이 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

경고

"name" 매개 변수는 새 일괄 처리를 만들 때마다 고유해야 합니다.

Scala 및 azdata

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

경고

일괄 처리 이름의 "name" 또는 "n" 매개 변수는 새 일괄 처리를 만들 때마다 고유해야 합니다.

Livy를 사용하는 Scala 및 curl

이 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

경고

일괄 처리 이름의 "name" 매개 변수는 새 일괄 처리를 만들 때마다 고유해야 합니다.

SQL 및 azdata

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

경고

일괄 처리 이름의 "name" 또는 "n" 매개 변수는 새 일괄 처리를 만들 때마다 고유해야 합니다.

Livy를 사용하는 SQL 및 curl

이 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

경고

"name" 매개 변수는 새 일괄 처리를 만들 때마다 고유해야 합니다.

Spark 작업 모니터링

azdata bdc spark batch 명령에서는 Spark 일괄 작업을 위한 관리 작업을 제공합니다.

실행 중인 모든 작업을 나열하려면 다음 명령을 실행합니다.

• azdata 명령은 다음을 수행합니다.

  azdata bdc spark batch list -o table
  

• Livy를 사용하는 curl 명령입니다.

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

지정된 ID를 사용하여 Spark 일괄 처리에 대한 정보를 가져오려면 다음 명령을 실행합니다. batch id는 spark batch create에서 반환됩니다.

• azdata 명령은 다음을 수행합니다.

  azdata bdc spark batch info --batch-id 0
  

• Livy를 사용하는 curl 명령입니다.

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

지정된 ID를 사용하여 Spark 일괄 처리에 대한 상태 정보를 가져오려면 다음 명령을 실행합니다.

• azdata 명령은 다음을 수행합니다.

  azdata bdc spark batch state --batch-id 0
  

• Livy를 사용하는 curl 명령입니다.

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

지정된 ID를 사용하여 Spark 일괄 처리에 대한 로그를 가져오려면 다음 명령을 실행합니다.

• azdata 명령은 다음을 수행합니다.

  azdata bdc spark batch log --batch-id 0
  

• Livy를 사용하는 curl 명령입니다.

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

다음 단계

Spark 코드 문제 해결에 대한 내용은 PySpark Notebook 문제 해결을 참조하세요.

전체 Spark 샘플 코드는 GitHub의 SQL Server 빅 데이터 클러스터 Spark 샘플에서 이용할 수 있습니다.

SQL Server 빅 데이터 클러스터 및 관련 시나리오에 대한 자세한 내용은 SQL Server 빅 데이터 클러스터를 참조하세요.