文档

离线Spark消费示例

更新时间:

本文介绍Spark如何访问SLS。

Spark RDD访问SLS

代码示例

## TestBatchLoghub.Scala

object TestBatchLoghub {
  def main(args: Array[String]): Unit = {
    if (args.length < 6) {
      System.err.println(
        """Usage: TestBatchLoghub <sls project> <sls logstore> <sls endpoint>
          |  <access key id> <access key secret> <output path> <start time> <end time=now>
        """.stripMargin)
      System.exit(1)
    }

    val loghubProject = args(0)
    val logStore = args(1)
    val endpoint = args(2)
    val accessKeyId = args(3)
    val accessKeySecret = args(4)
    val outputPath = args(5)
    val startTime = args(6).toLong

    val sc = new SparkContext(new SparkConf().setAppName("test batch loghub"))
    var rdd:JavaRDD[String] = null
    if (args.length > 7) {
      rdd = LoghubUtils.createRDD(sc, loghubProject, logStore, accessKeyId, accessKeySecret, endpoint, startTime, args(7).toLong)
    } else {
      rdd = LoghubUtils.createRDD(sc, loghubProject, logStore, accessKeyId, accessKeySecret, endpoint, startTime)
    }

    rdd.saveAsTextFile(outputPath)
  }
}
说明

Maven pom文件可以参见aliyun-emapreduce-demo

编译运行

说明

运行代码示例前必须先配置环境变量。关于如何配置环境变量,请参见配置环境变量

## 编译命令
mvn clean package -DskipTests

## 编译完后,作业JAR包位于target/shaded/下。

## 提交执行
spark-submit --master yarn-cluster --executor-cores 2 --executor-memory 1g --driver-memory 1g --num-executors 2 --class x.x.x.TestBatchLoghub xxx.jar <sls project> <sls logstore> <sls endpoint> $ALIBABA_CLOUD_ACCESS_KEY_ID $ALIBABA_CLOUD_ACCESS_KEY_SECRET <output path> <start time> [<end time=now>]
重要
  • x.x.x.TestBatchLoghubxxx.jar需要替换成真实的类路径和包路径。

  • 作业资源需要根据实际数据规模和实际集群规模调整,如果集群太小,直接运行以上命令可能无法执行。

spark-sql访问SLS

访问命令

spark-sql --jars /opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/* \
  --hiveconf accessKeyId=$ALIBABA_CLOUD_ACCESS_KEY_ID \
  --hiveconf accessKeySecret=$ALIBABA_CLOUD_ACCESS_KEY_SECRET
说明

/opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/*中包含LogHub DataSource类型。如果您EMR集群使用的是Spark2,则应修改上面命令中的spark3应该换成spark2

如果您希望在本地电脑的开发环境中使用Spark3依赖SLS,类似于Spark2的操作方式,可以按照以下步骤操作:

  1. 下载集群/opt/apps/SPARK-EXTENSION/spark-extension-current/spark3-emrsdk/emr-datasources_shaded_2.12目录下的内容到本地。

  2. 使用Maven将JAR包安装到本地。

    mvn install:install-file -DgroupId=com.aliyun.emr -DartifactId=emr-datasources_shaded_2.12 -Dversion=3.0.2 -Dpackaging=jar -Dfile=/Users/zhongqiang.czq/Downloads/tempory/emr-datasources_shaded_2.12-3.0.2.jar
  3. 在pom文件中添加以下依赖项。

    <dependency>
      <groupId>com.aliyun.emr</groupId>
      <artifactId>emr-datasources_shaded_2.12</artifactId>
      <version>3.0.2</version>
    </dependency>

建表和读取数据示例

create table test_sls
using loghub
  options(endpoint='cn-hangzhou-intranet.log.aliyuncs.com',
          access.key.id='${hiveconf:accessKeyId}',
          access.key.secret='${hiveconf:accessKeySecret}',
          sls.project='test_project',
          sls.store='test_store',
          startingoffsets='earliest'
);

select * from test_sls;

配置环境变量

配置环境变量ALIBABA_CLOUD_ACCESS_KEY_IDALIBABA_CLOUD_ACCESS_KEY_SECRET

重要
  • 阿里云账号AccessKey拥有所有API的访问权限,建议您使用RAM用户进行API访问或日常运维,具体操作,请参见创建RAM用户

  • 请不要将AccessKey ID和AccessKey Secret保存到工程代码里,否则可能导致AccessKey泄露,威胁您账号下所有资源的安全。

  • LinuxmacOS系统配置方法:

    执行以下命令配置环境变量。

    其中, <access_key_id>需替换为您RAM用户的AccessKey ID,<access_key_secret>替换为您RAM用户的AccessKey Secret。

    export ALIBABA_CLOUD_ACCESS_KEY_ID=<access_key_id>
    export ALIBABA_CLOUD_ACCESS_KEY_SECRET=<access_key_secret>
  • Windows系统配置方法

    1. 新建环境变量文件,添加环境变量ALIBABA_CLOUD_ACCESS_KEY_IDALIBABA_CLOUD_ACCESS_KEY_SECRET,并写入已准备好的AccessKey ID和AccessKey Secret。

    2. 重启Windows系统生效。

相关文档

Spark访问Kafka:Structured Streaming + Kafka Integration Guide