Spark作业读写OSS数据

使用Hadoop OSS SDK

以Spark 3.5.5版本和Hadoop OSS SDK 3.3.4版本为例，创建如下示例Dockerfile文件。

ARG SPARK_IMAGE=spark:3.5.5

FROM ${SPARK_IMAGE}

# Add dependency for Hadoop Aliyun OSS support
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aliyun/3.3.4/hadoop-aliyun-3.3.4.jar ${SPARK_HOME}/jars
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/com/aliyun/oss/aliyun-sdk-oss/3.17.4/aliyun-sdk-oss-3.17.4.jar ${SPARK_HOME}/jars
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/org/jdom/jdom2/2.0.6.1/jdom2-2.0.6.1.jar ${SPARK_HOME}/jars

使用Hadoop S3 SDK

以Spark 3.5.5版本和Hadoop S3 SDK 3.3.4版本为例，创建如下示例Dockerfile文件。

ARG SPARK_IMAGE=spark:3.5.5

FROM ${SPARK_IMAGE}

# Add dependency for Hadoop AWS S3 support
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.3.4/hadoop-aws-3.3.4.jar ${SPARK_HOME}/jars
ADD --chown=spark:spark --chmod=644 https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.12.367/aws-java-sdk-bundle-1.12.367.jar ${SPARK_HOME}/jars

使用JindoSDK

以Spark 3.5.5版本和JindoSDK 6.8.0版本为例，创建如下示例Dockerfile文件。

ARG SPARK_IMAGE=spark:3.5.5

FROM ${SPARK_IMAGE}

# Add dependency for JindoSDK support
ADD --chown=spark:spark --chmod=644 https://jindodata-binary.oss-cn-shanghai.aliyuncs.com/mvn-repo/com/aliyun/jindodata/jindo-core/6.8.0/jindo-core-6.8.0.jar ${SPARK_HOME}/jars
ADD --chown=spark:spark --chmod=644 https://jindodata-binary.oss-cn-shanghai.aliyuncs.com/mvn-repo/com/aliyun/jindodata/jindo-sdk/6.8.0/jindo-sdk-6.8.0.jar ${SPARK_HOME}/jars

使用Hadoop OSS SDK

1. 根据以下内容创建Secret清单文件，并保存为 spark-oss-secret.yaml。

   apiVersion: v1
   kind: Secret
   metadata:
     name: spark-oss-secret
     namespace: default
   stringData:
     # 需将 <ACCESS_KEY_ID> 替换成阿里云 AccessKey ID。
     OSS_ACCESS_KEY_ID: <ACCESS_KEY_ID>
     # 需将 <ACCESS_KEY_SECRET> 替换成阿里云 AccessKey Secret。
     OSS_ACCESS_KEY_SECRET: <ACCESS_KEY_SECRET>

2. 执行如下命令创建Secret。

   kubectl apply -f spark-oss-secret.yaml

   预期输出如下。

   secret/spark-oss-secret created

使用Hadoop S3 SDK

1. 根据以下内容创建Secret清单文件，并保存为 spark-s3-secret.yaml。

   apiVersion: v1
   kind: Secret
   metadata:
     name: spark-s3-secret
     namespace: default
   stringData:
     # 需将 <ACCESS_KEY_ID> 替换成阿里云 AccessKey ID。
     AWS_ACCESS_KEY_ID: <ACCESS_KEY_ID> 
     # 需将 <ACCESS_KEY_SECRET> 替换成阿里云 AccessKey Secret。
     AWS_SECRET_ACCESS_KEY: <ACCESS_KEY_SECRET>

2. 执行如下命令创建Secret。

   kubectl apply -f spark-s3-secret.yaml

   预期输出如下。

   secret/spark-s3-secret created

使用JindoSDK

1. 根据以下内容创建Secret清单文件，并保存为 spark-oss-secret.yaml。

   apiVersion: v1
   kind: Secret
   metadata:
     name: spark-oss-secret
     namespace: default
   stringData:
     # 需将 <ACCESS_KEY_ID> 替换成阿里云 AccessKey ID。
     OSS_ACCESS_KEY_ID: <ACCESS_KEY_ID>
     # 需将 <ACCESS_KEY_SECRET> 替换成阿里云 AccessKey Secret。
     OSS_ACCESS_KEY_SECRET: <ACCESS_KEY_SECRET>

2. 执行如下命令创建Secret。

   kubectl apply -f spark-oss-secret.yaml

   预期输出如下。

   secret/spark-oss-secret created

使用Hadoop OSS SDK

创建如下SparkApplication清单文件并保存为spark-pagerank.yaml。关于OSS完整的配置参数列表，请参见Hadoop-Aliyun module。

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pagerank
  namespace: default
spec:
  type: Scala
  mode: cluster
  # 需将 <SPARK_IMAGE> 替换成步骤二中构建得到的 Spark 容器镜像。
  image: <SPARK_IMAGE>
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.5.jar
  mainClass: org.apache.spark.examples.SparkPageRank
  arguments:
  - oss://<OSS_BUCKET>/data/pagerank_dataset.txt           # 指定输入测试数据集，将<OSS_BUCKET>替换成OSS Bucket名称。
  - "10"                                                   # 迭代次数。
  sparkVersion: 3.5.5
  hadoopConf:
    fs.AbstractFileSystem.oss.impl: org.apache.hadoop.fs.aliyun.oss.OSS
    fs.oss.impl: org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem
    # OSS 访问端点，需将 <OSS_ENDPOINT> 替换成您的 OSS 访问端点，
    # 例如，北京地域 OSS 的内网访问端点为 oss-cn-beijing-internal.aliyuncs.com。 
    fs.oss.endpoint: <OSS_ENDPOINT>
    fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider
  driver:
    cores: 1
    coreLimit: 1200m
    memory: 512m
    envFrom:
    - secretRef:
        name: spark-oss-secret               #指定访问OSS的安全凭据。
    serviceAccount: spark-operator-spark
  executor:
    instances: 2
    cores: 1
    coreLimit: "2"
    memory: 8g
    envFrom:
    - secretRef: 
        name: spark-oss-secret               #指定访问OSS的安全凭据。
  restartPolicy:
    type: Never

使用Hadoop S3 SDK

创建如下SparkApplication清单文件并保存为spark-pagerank.yaml。关于S3完整的配置参数列表，请参见Hadoop-AWS module。

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pagerank
  namespace: default
spec:
  type: Scala
  mode: cluster
  # 需将 <SPARK_IMAGE> 替换成步骤二中构建得到的 Spark 容器镜像。
  image: <SPARK_IMAGE>
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.5.jar
  mainClass: org.apache.spark.examples.SparkPageRank
  arguments:
  - s3a://<OSS_BUCKET>/data/pagerank_dataset.txt           # 指定输入测试数据集，将<OSS_BUCKET>替换成OSS Bucket名称。
  - "10"                                                   # 迭代次数。
  sparkVersion: 3.5.5
  hadoopConf:
    fs.s3a.impl: org.apache.hadoop.fs.s3a.S3AFileSystem
    # OSS 访问端点，需将 <OSS_ENDPOINT> 替换成您的 OSS 访问端点，
    # 例如，北京地域 OSS 的内网访问端点为 oss-cn-beijing-internal.aliyuncs.com。
    fs.s3a.endpoint: <OSS_ENDPOINT>
    # OSS 访问端点所在地域，例如北京地域为 cn-beijing。
    fs.s3a.endpoint.region: <OSS_REGION>
  driver:
    cores: 1
    coreLimit: 1200m
    memory: 512m
    envFrom:
    - secretRef:
        name: spark-s3-secret               #指定访问OSS的安全凭据。
    serviceAccount: spark-operator-spark
  executor:
    instances: 2
    cores: 1
    coreLimit: "2"
    memory: 8g
    envFrom:
    - secretRef: 
        name: spark-s3-secret               #指定访问OSS的安全凭据。
  restartPolicy:
    type: Never

使用JindoSDK

创建如下SparkApplication清单文件并保存为spark-pagerank.yaml。

apiVersion: sparkoperator.k8s.io/v1beta2
kind: SparkApplication
metadata:
  name: spark-pagerank
  namespace: default
spec:
  type: Scala
  mode: cluster
  # 需将 <SPARK_IMAGE> 替换成步骤二中构建得到的 Spark 容器镜像。
  image: <SPARK_IMAGE>
  mainApplicationFile: local:///opt/spark/examples/jars/spark-examples_2.12-3.5.5.jar
  mainClass: org.apache.spark.examples.SparkPageRank
  arguments:
  - oss://<OSS_BUCKET>/data/pagerank_dataset.txt    # 指定输入测试数据集，将<OSS_BUCKET>替换成OSS Bucket名称。
  - "10"                                            # 迭代次数。
  sparkVersion: 3.5.5
  hadoopConf:
    fs.AbstractFileSystem.oss.impl: com.aliyun.jindodata.oss.JindoOSS
    fs.oss.impl: com.aliyun.jindodata.oss.JindoOssFileSystem
    fs.oss.endpoint: <OSS_ENDPOINT>                 # OSS访问端点。例如北京地区OSS的内网访问地址为oss-cn-beijing-internal.aliyuncs.com。
    fs.oss.credentials.provider: com.aliyun.jindodata.oss.auth.EnvironmentVariableCredentialsProvider
  driver:
    cores: 1
    coreLimit: 1200m
    memory: 512m
    serviceAccount: spark-operator-spark
    envFrom:
    - secretRef:
        name: spark-oss-secret                    #指定访问OSS的安全凭据。
  executor: 
    instances: 2
    cores: 1
    coreLimit: "2"
    memory: 8g
    envFrom:
    - secretRef:
        name: spark-oss-secret                    #指定访问OSS的安全凭据。
  restartPolicy:
    type: Never

使用Hadoop OSS SDK

kubectl delete -f spark-oss-secret.yaml

使用Hadoop S3 SDK

kubectl delete -f spark-s3-secret.yaml

使用JindoSDK

kubectl delete -f spark-oss-secret.yaml