Spark

Spark

Apache Spark를 이용해 Apache Pinot에 배치 수집을 수행하는 방법을 다루는 페이지예요.

출처: Spark

본문

Pinot은 Apache Spark 3.x를 프로세서로 사용해 세그먼트 파일을 생성하고 데이터베이스에 푸시하는 것을 지원해요. Pinot 배포판에는 파일을 처리하고 Pinot으로 변환·업로드하는 Spark 코드가 번들되어 있어요.

Spark를 설정하려면 다음 중 하나를 수행하세요:

  • Spark-Pinot 커넥터를 사용. 자세한 내용은 ReadMe 참고.
  • 아래 지침을 따르세요.

로컬 설치 가이드를 따라 소스에서 Pinot을 빌드할 수 있어요. 결과 JAR 파일은 pinot/target/pinot-all-${PINOT_VERSION}-jar-with-dependencies.jar에서 찾을 수 있어요.

소스에서 Pinot을 빌드한다면 -Pbuild-shaded-jar로 build-shaded-jar jar 프로파일을 사용하는 것을 고려해 보세요. Pinot은 jar에 Spark를 번들하지 않지만 일부 Hadoop 라이브러리는 번들해요.

다음으로 잡 스펙의 실행 설정을 다음과 같이 바꿔야 해요:

# executionFrameworkSpec: Defines ingestion jobs to be running.
executionFrameworkSpec:

  # name: execution framework name
  name: 'spark'

  # segmentGenerationJobRunnerClassName: class name implements org.apache.pinot.spi.ingestion.batch.runner.IngestionJobRunner interface.
  segmentGenerationJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.spark3.SparkSegmentGenerationJobRunner'

  # segmentTarPushJobRunnerClassName: class name implements org.apache.pinot.spi.ingestion.batch.runner.IngestionJobRunner interface.
  segmentTarPushJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.spark3.SparkSegmentTarPushJobRunner'

  # segmentUriPushJobRunnerClassName: class name implements org.apache.pinot.spi.ingestion.batch.runner.IngestionJobRunner interface.
  segmentUriPushJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.spark3.SparkSegmentUriPushJobRunner'

  #segmentMetadataPushJobRunnerClassName: class name implements org.apache.pinot.spi.ingestion.batch.runner.IngestionJobRunner interface
  segmentMetadataPushJobRunnerClassName: 'org.apache.pinot.plugin.ingestion.batch.spark3.SparkSegmentMetadataPushJobRunner'

  # extraConfigs: extra configs for execution framework.
  extraConfigs:

    # stagingDir is used in distributed filesystem to host all the segments then move this directory entirely to output directory.
    stagingDir: your/local/dir/staging

Spark 잡 스펙은 includeFileNamePattern과 excludeFileNamePattern으로 입력 경로를 필터링할 수 있어요. 두 속성 모두 Java NIO glob: 및 regex: 패턴을 받아요; 동작 예시와 경로 정규화 세부 사항은 File name patterns 참고.

필요한 jar와 plugins.dir

pinot-all-*-jar-with-dependencies.jar는 Spark 또는 Hadoop 배치 수집 플러그인을 번들하지 않아요. 그것들은 바이너리 배포판의 plugins-external/ 아래에 있어요 (pinot-assembly.xml 참고). 그것들을 Spark 클래스패스에 넣고 plugins.dir이 그것들을 포함하는 디렉터리를 가리켜야 해요.

Spark 수집을 실행하려면 다음이 필요해요:

  • pinot-all jar — 패키지의 lib/ 아래
  • pinot-batch-ingestion-spark-3 셰이딩 플러그인 jar — plugins-external/pinot-batch-ingestion/pinot-batch-ingestion-spark-3/ 아래
  • 잡이 사용하는 기타 플러그인(레코드 리더, 파일 시스템) — plugins/ 아래 (예: pinot-avro, pinot-parquet, pinot-s3)

plugins.dir은 세미콜론으로 구분된 디렉터리 목록을 받아요. 레코드 리더와 Spark 배치 러너가 올바르게 로드되도록 plugins와 plugins-external을 모두 포함하세요:

-Dplugins.dir=${PINOT_DISTRIBUTION_DIR}/plugins;${PINOT_DISTRIBUTION_DIR}/plugins-external

Spark 배치 플러그인과 pinot-all을 spark.driver.extraClassPath / spark.executor.extraClassPath로 드라이버(및 실행자) 클래스패스에 넣으세요:

spark.driver.extraClassPath =>
${PINOT_DISTRIBUTION_DIR}/plugins-external/pinot-batch-ingestion/pinot-batch-ingestion-spark-3/pinot-batch-ingestion-spark-3-${PINOT_VERSION}-shaded.jar:${PINOT_DISTRIBUTION_DIR}/lib/pinot-all-${PINOT_VERSION}-jar-with-dependencies.jar

완전한 로컬 모드 spark-submit 명령:

export PINOT_VERSION=1.4.0 #set to the Pinot version you have installed
export PINOT_DISTRIBUTION_DIR=/path/to/apache-pinot-${PINOT_VERSION}-bin

SPARK_BATCH_PLUGIN=${PINOT_DISTRIBUTION_DIR}/plugins-external/pinot-batch-ingestion/pinot-batch-ingestion-spark-3/pinot-batch-ingestion-spark-3-${PINOT_VERSION}-shaded.jar
PINOT_ALL_JAR=${PINOT_DISTRIBUTION_DIR}/lib/pinot-all-${PINOT_VERSION}-jar-with-dependencies.jar

spark-submit \
  --class org.apache.pinot.tools.admin.command.LaunchDataIngestionJobCommand \
  --master local --deploy-mode client \
  --conf "spark.driver.extraJavaOptions=-Dplugins.dir=${PINOT_DISTRIBUTION_DIR}/plugins;${PINOT_DISTRIBUTION_DIR}/plugins-external" \
  --conf "spark.driver.extraClassPath=${SPARK_BATCH_PLUGIN}:${PINOT_ALL_JAR}" \
  --conf "spark.executor.extraClassPath=${SPARK_BATCH_PLUGIN}:${PINOT_ALL_JAR}" \
  local://${PINOT_ALL_JAR} \
  -jobSpecFile /path/to/spark_job_spec.yaml

환경 변수 PINOT_ROOT_DIR과 PINOT_VERSION이 올바르게 설정되었는지 확인하세요.

참고: 프로덕션 환경에서는 master를 yarn으로, deploy-mode를 cluster로 변경해야 해요.

{% hint style="info" %} spark-core 의존성은 0.10.0 릴리스부터 Pinot jar에 포함되지 않습니다. 런타임 문제가 발생하면 Spark 환경이 그 의존성을 제공하는지 확인하거나, 일치하는 Spark 프로파일로 소스에서 빌드하세요. {% endhint %}

YARN에서 실행하기

아래 예제는 plugins.dir이 참조하는 압축 해제된 Pinot 배포판을 Spark 드라이버가 읽을 수 있도록 YARN에서 client 배포 모드를 사용해요. 실행 전에:

  • -DuseProvidedHadoop 옵션으로 소스에서 Pinot을 빌드
  • 압축 해제된 Pinot 배포판을 제출 호스트에 보관
  • 수집 스펙 YAML을 --files로 접근 가능한 S3, HDFS 또는 다른 위치로 복사
  • --jars로 Spark 배치 플러그인과 pinot-all을 추가해 Spark가 실행자에 분배하도록 함
  • 드라이버와 실행자 클래스패스를 분배된 jar 이름으로 지정

YARN cluster 배포 모드에서는 드라이버가 원격에서 실행돼요. plugins와 plugins-external 디렉터리를 별도로 분배하고 압축 해제한 다음, 원격 드라이버 컨테이너에 존재하는 경로로 plugins.dir을 설정해야 해요.

예시

spark-submit \
  --class org.apache.pinot.tools.admin.command.LaunchDataIngestionJobCommand \
  --master yarn --deploy-mode client \
  --conf "spark.driver.extraJavaOptions=-Dplugins.dir=${PINOT_DISTRIBUTION_DIR}/plugins;${PINOT_DISTRIBUTION_DIR}/plugins-external" \
  --conf "spark.driver.extraClassPath=pinot-batch-ingestion-spark-3-${PINOT_VERSION}-shaded.jar:pinot-all-${PINOT_VERSION}-jar-with-dependencies.jar" \
  --conf "spark.executor.extraClassPath=pinot-batch-ingestion-spark-3-${PINOT_VERSION}-shaded.jar:pinot-all-${PINOT_VERSION}-jar-with-dependencies.jar" \
  --jars "${PINOT_DISTRIBUTION_DIR}/plugins-external/pinot-batch-ingestion/pinot-batch-ingestion-spark-3/pinot-batch-ingestion-spark-3-${PINOT_VERSION}-shaded.jar,${PINOT_DISTRIBUTION_DIR}/lib/pinot-all-${PINOT_VERSION}-jar-with-dependencies.jar" \
  --files s3://path/to/spark_job_spec.yaml \
  ${PINOT_DISTRIBUTION_DIR}/lib/pinot-all-${PINOT_VERSION}-jar-with-dependencies.jar \
  -jobSpecFile spark_job_spec.yaml

FAQ

Q - 다음 예외가 발생해요 - Class has been compiled by a more recent version of the Java Runtime (class file version 55.0), this version of the Java Runtime only recognizes class file versions up to 52.0

0.8.0 릴리스부터 Pinot 바이너리는 JDK 11로 컴파일돼요. Spark를 Hadoop 2.7+와 함께 사용한다면 Java8 버전의 Pinot을 사용해야 해요. 현재는 소스에서 jdk 8 버전을 빌드해야 해요.

Q - pinot-batch-ingestion-spark jar를 찾을 수 없어요.

Pinot 0.10.0부터 Spark 배치 플러그인은 바이너리 배포판의 plugins-external/pinot-batch-ingestion/ 아래에 있어요 (이전 버전에서는 plugins/ 아래에 있었음). 현재 Spark 3 빌드는 pinot-batch-ingestion-spark-3-*-shaded.jar를 제공해요.

Q - Spark가 jar를 찾지 못해 java.nio.file.NoSuchFileException이 발생해요

이것은 Spark 잡의 클래스패스가 제대로 설정되지 않았다는 뜻이에요. Yarn이나 k8s 같은 분산 환경에서 Spark를 실행한다면 spark.driver.extraClassPath와 spark.executor.extraClassPath가 모두 설정되었는지 확인하세요. 또한 driver.extraClassPath의 jar를 spark-submit의 --jars 인자에 추가해 Spark가 그 jar를 클러스터의 모든 노드에 분배할 수 있게 해야 해요. jar를 실행할 때도 파일 경로에 적절한 스킴을 제공해야 해요. 이 문서에서는 local://를 사용했지만 클러스터 설정에 따라 달라질 수 있어요.

Q - 세그먼트를 푸시하는 동안 Spark 잡이 실패해요

잡 스펙 yaml 파일의 controllerURI가 잘못 구성됐을 수 있어요. controllerURI가 올바르다면, YARN이나 k8s 클러스터의 모든 노드에서 접근 가능한지 확인하세요.

Q - 수집 중 데이터가 덮어써져요

tableConfig에서 segmentPushType을 APPEND로 설정하세요.

이미 APPEND로 설정되어 있다면 테이블 설정에 timeColumnName이 없어서일 가능성이 커요. 시간 컬럼을 제공할 수 없다면 수집 스펙에서 세그먼트 이름 생성 설정을 사용하세요. 일반적으로 inputFile 세그먼트 이름 생성기를 사용하면 문제가 해결돼요.

Q - java.lang.RuntimeException: java.io.IOException: Failed to create directory: pinot-plugins-dir-0/plugins/*이 발생해요

플러그인 jar가 extraClassPath와 --jars를 통해 이미 Spark 클래스패스에 있다면 spark.driver.extraJavaOptions에서 -Dplugins.dir=...을 제거해 해결될 수 있어요. plugins.dir은 드라이버에 존재하는 실제 로컬 경로(배포판 아래의 plugins와 plugins-external)를 가리키는 것을 선호해요.

Q - Class not found: 예외 발생 (예: SparkSegmentGenerationJobRunner)

Spark/Hadoop 배치 러너는 pinot-all 안에 없어요. 다음을 확인하세요:

  1. 드라이버와 실행자의 extraClassPath가 plugins-external/pinot-batch-ingestion/pinot-batch-ingestion-spark-3/ 아래의 셰이딩 jar를 포함
  2. 같은 jar가 클러스터 모드에서 --jars에 나열
  3. plugins.dir이 ${PINOT_DISTRIBUTION_DIR}/plugins-external을 포함 (plugins/의 레코드 리더/파일시스템 플러그인도 필요하다면 plugins와 세미콜론 구분)
  4. 잡 스펙이 Spark 3 패키지(org.apache.pinot.plugin.ingestion.batch.spark3.*)를 사용

더 알아보기 (Learn more)