通过EMR Spark Streaming实时读取Tablestore数据

本文将介绍如何在E-MapReduce中实时流式的处理Tablestore中的数据。

场景设计

随着互联网的发展，企业中积累的数据越来越多，数据的背后隐藏着巨大的价值，在双十一这样的节日中，电子商务企业都会在大屏幕上实时显示订单总量，由于订单总量巨大，不可能每隔一秒就到数据库中进行一次SQL统计，此时就需要用到流计算，而传统的方法都是需要借助Kafka消息队列来做流式计算，数据订单需要写入数据库与Kafka中，Spark Streaming 消费来自Kafka中的订单信息。
而本文使用的Tablestore数据库可以直接利用它的通道服务功能，供Spark Streaming流式消费，进而计算订单的数量及金额，简化了整个流程，具体如下图所示


本文将介绍一个简单的demo，流式统计Tablestore数据表中字段出现的个数。

前提条件

确保将Tablestore实例部署在E-MapReduce集群相同的VPC环境下

准备工作

• 登录阿里云E-MapReduce控制台
• 创建Hadoop集群

步骤一 创建Tablestore数据源表

详细开通步骤请参考官方文档，本文demo中所创建出来的表名为SmallTarget，表的Schema如下图所示，该表有PKString和PkInt两个主键，类型分别为String和Interger。


为表SmallTarget建立一个增量通道，如下图所示，通道列表里面会显示该通道的名字、ID以及类型。

技术注解:
通道服务（Tunnel Service）是基于Tablestore数据接口之上的全增量一体化服务，包含三种通道类型：

• 全量：对数据表中历史存量数据消费处理
• 增量：对数据表中新增数据消费处理
• 全量加增量：先对数据表总历史存量数据消费，之后对新增数据消费

通道服务的详细介绍可查询> Tablestore官网文档。

步骤二 获取相关jar包并上传到hadoop集群

• 获取环境依赖的JAR包。

• 在集群管理页面，单击已创建的Hadoop集群的集群ID ，进入集群与服务管理页面。
• 在左侧导航树中选择主机列表，然后在右侧查看Hadoop集群中emr-header-1主机的IP信息。
• 在SSH客户端中新建一个命令窗口，登录Hadoop集群的emr-header-1主机。
• 上传所有JAR包到emr-header-1节点的某个目录下。

步骤三 运行Spark Streaming作业

1. 以一个基于emr demo修改的WordCount为样例，编译生成JAR包，JAR包需要上传到Hadoop集群的emr-header-1主机中(参见步骤二)，本例测试用的JAR包地址，完整的WordCount代码参见附录(后续会合到emr demo项目中)。
2. 该样例以Tablestore表作为数据源，统计的是主键PkString。SmallTarget数据表中数据初始时为空，如下图所示。
   

3. 启动spark streaming，开启一个统计SmallTarget表中通道zengliang(通道名)实时消费数据的一个监听程序。

spark-submit --class com.aliyun.emr.example.spark.sql.streaming.StructuredTableStoreWordCount  --jars emr-tablestore-X.X.X-SNAPSHOT.jar,tablestore-X.X.X-jar-with-dependencies.jar examples-X.X.X-shaded.jar <instance> <tableName> <tunnelId> <accessKeyId> <accessKeySecret> <endPoint> <maxOffsetsPerChannel>
-------------------------------------------
Batch: 0
-------------------------------------------
+-----+-----+
|value|count|
+-----+-----+
+-----+-----+

各个参数的详细说明如下表所示。

向SmallTarget数据表插入数据，spark streaming会实时更新,如下面两张图所示。

Catalog字段说明

Catalog为一个Json字符串，"columns"里面指定的是需要读取的列的自定义配置，这些列的键值对最终会被TableStore Source所输出，如下图所示，会读取PkString, PkInt和col1三列，其类型分别为string, long和string。

{
  "columns": {
    "PkString": {
      "col": "PkString",
      "type": "string"
    },
    "PkInt": {
      "col": "PkInt",
      "type": "long"
    },
    "col1": {
      "col": "col1",
      "type": "string"
    }
  }
}

预定义列说明

除去Catalog字段中的用户自定义列之外，Tablestore Source的输出默认还会加上一些预定义列的值，以供下游系统灵活的使用，预定义字段说明如下表所示：

附录

package com.aliyun.emr.example.spark.sql.streaming

import java.util.UUID

import org.apache.spark.sql.SparkSession

object StructuredTableStoreWordCount {
  def main(args: Array[String]): Unit = {
    if (args.length < 7) {
      System.err.println(
        "Usage: StructuredTableStoreWordCount <ots-instanceName>" +
          "<ots-tableName> <ots-tunnelId> <access-key-id> <access-key-secret> <ots-endpoint>" +
          "<max-offsets-per-channel> [<checkpoint-location>]"
      )
    }

    val Array(
    instanceName,
    tableName,
    tunnelId,
    accessKeyId,
    accessKeySecret,
    endpoint,
    maxOffsetsPerChannel,
    _*
    ) = args

    System.out.println(args.toSeq.toString)

    val checkpointLocation =
      if (args.length > 7) args(7) else "/tmp/temporary-" + UUID.randomUUID.toString

    val spark = SparkSession.builder.appName("TableStoreWordCount").master("local[5]").getOrCreate()

    spark.sparkContext.setLogLevel("WARN")

    import spark.implicits._

    val lines = spark.readStream
      .format("tablestore")
      .option("instance.name", instanceName)
      .option("table.name", tableName)
      .option("tunnel.id", tunnelId)
      .option("endpoint", endpoint)
      .option("access.key.id", accessKeyId)
      .option("access.key.secret", accessKeySecret)
      .option("maxOffsetsPerChannel", maxOffsetsPerChannel) // default 10000
      .option(
        "catalog",
        "{\"columns\":{\"PkString\":{\"col\":\"PkString\",\"type\":\"string\"},\"PkInt\":{\"col\":\"PkInt\",\"type\":\"long\"}," +
          "\"col1\":{\"col\":\"col1\",\"type\":\"string\"}}}"
      )
      .load()
      .selectExpr("PkString", "PkInt", "__ots_record_type__")
      .as[(String, Long, String)]

    val wordCounts = lines
      .flatMap(line => {
        System.out.println(s"wordCount line: ${line}")
        line._1.split(" ")
      })
      .groupBy("value")
      .count()

    val query = wordCounts.writeStream
      .outputMode("complete")
      .format("console")
      .option("checkpointLocation", checkpointLocation)
      .start()

    query.awaitTermination()
  }
}

写在最后

本篇文章我们介绍了如何在E-MapReduce中实时流式的处理Tablestore中的数据，如果对基于Tablestore的大数据存储分析感兴趣的朋友可以加入我们的技术交流群（钉钉：23307953 或者11789671），来与我们一起探讨。