使用EMR DataFrame 批处理 Tablestore-阿里云开发者社区

开发者社区> 开发与运维> 正文
登录阅读全文

使用EMR DataFrame 批处理 Tablestore

简介: 使用Spark的DataFrame方式访问表格存储,并在本地和集群上分别进行运行调试。 ## 前提条件 - 了解Spark访问表格存储的依赖包,并在使用时通过maven方式引入项目中。 - Spark相关:spark-core、spark-sql、spark-hive - Spark Tablestore connector:emr-tablestore-.jar

使用Spark的DataFrame方式访问表格存储,并在本地和集群上分别进行运行调试。

前提条件

  • 了解Spark访问表格存储的依赖包,并在使用时通过maven方式引入项目中。

    • Spark相关:spark-core、spark-sql、spark-hive

    • Spark Tablestore connector:emr-tablestore-.jar

    • Tablestore Java SDK:tablestore--jar-with-dependencies.jar

    其中表示相应依赖包的版本号,请以实际为准。

  • 已在表格存储侧创建数据表,详情请参见概述

快速开始

通过项目样例了解快速使用批计算的操作。

  1. 从GitHub下载项目样例的源码,具体下载路径请参见TableStoreSparkDemo。项目中包含完整的依赖和使用样例,具体的依赖请参见项目中的pom文件。

  2. 阅读TableStoreSparkDemo项目的README文档,并安装最新版的Spark Tablestore connector和Tablestore Java SDK到本地maven库。

    说明:Spark Tablestore connector正式版发布以月为周期,目前最新版尚未正式发布,请先使用项目附带的预览版,正式发布后,本文也会进行更新,敬请期待。预览版和正式版只是版本号的区别,相互兼容,业务代码逻辑无需改动。

  3. 修改Sample代码
    以TableStoreBatchSample为例,对此示例代码的核心代码说明如下:

    • format("tablestore")表示使用ServiceLoader方式加载Spark Tablestore connector,具体配置请参见项目中的META-INF.services。

    • instanceName、tableName、endpoint、accessKeyId、accessKeySecret分别表示表格存储的实例名称、数据表名称、实例endpoint、阿里云账号的AccessKey ID和AccessKey Secret。

    • catalog是一个json串,包含字段名和类型,如下示例中的数据表有salt(Long类型)、UserId(String类型)、OrderId(String类型)、price(Double类型)和timestamp(Long类型)五个字段。最新版本中支持使用Schema方式替换catalog的配置,请根据实际选择。

    • split.size.mbs表示每个Split的切分大小,默认值为100,单位为MB,可不配置。此值越小产生的Split会越多,对应Spark的Task也会越多。

    val df = sparkSession.read
    .format("tablestore")
    .option("instance.name", instanceName)
    .option("table.name", tableName)
    .option("endpoint", endpoint)
    .option("access.key.id", accessKeyId)
    .option("access.key.secret", accessKeySecret)
    .option("split.size.mbs", 100)
    .option("catalog", dataCatalog)
    // 最新版本支持使用Schema方式替换catalog的配置。
    //.schema("salt LONG, UserId STRING, OrderId STRING, price DOUBLE, timestamp LONG")
    .load()
    
    val dataCatalog: String =
      s"""
         |{"columns": {
         |    "salt": {"type":"long"},
         |    "UserId": {"type":"string"},
         |    "OrderId": {"type":"string"},
         |    "price": {"type":"double"},
         |    "timestamp": {"type":"long"}
         | }
         |}""".stripMargin

运行调试

根据需求修改示例代码后,可在本地或者通过Spark集群进行运行调试。以TableStoreBatchSample为例说明调试过程。

本地调试

以Intellij IDEA为例说明。

说明:本文测试使用的环境为Spark 2.4.3、Scala 2.11.7和Java SE Development Kit 8,如果使用中遇到问题,请联系表格存储技术支持。

  1. 在系统参数中,配置实例名称、数据表名称、实例endpoint、阿里云账号的AccessKey ID和AccessKey Secret等参数。您也可以自定义参数的加载方式。

  2. 选择include dependencies with "provided" scope,单击OK。

  3. 运行示例代码程序。

    image.png

通过Spark集群调试

以spark-submit方式为例说明。示例代码中的master默认为local[*],在Spark集群上运行时可以去掉,使用spark-submit参数传入。

  1. 执行mvn -U clean package命令打包,包的路径为target/tablestore-spark-demo-1.0-SNAPSHOT-jar-with-dependencies.jar。

  2. 上传包到Spark集群的Driver节点,并使用spark-submit提交任务。

    spark-submit --class com.aliyun.tablestore.spark.demo.batch.TableStoreBatchSample --master yarn tablestore-spark-demo-1.0-SNAPSHOT-jar-with-dependencies.jar <ots-instanceName> <ots-tableName> <access-key-id> <access-key-secret> <ots-endpoint>
    

    image.png

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

分享:
开发与运维
使用钉钉扫一扫加入圈子
+ 订阅

集结各类场景实战经验,助你开发运维畅行无忧

其他文章