简介

本文将介绍如何利用阿里云的sls插件功能和emr来进行mysql binlog的准实时传输

基本架构

rds -> sls -> spark streaming -> spark hdfs

主要包含3个链路：

1. 怎么把rds的binlog收集到sls

2.怎么通过spark streaming将sls中的日志读取出来，进行分析

3.怎么把2中读取和处理过的日志，保存到spark hdfs中

环境准备

需要mysql类型数据库、开通sls服务并添加对应的project和logstore、然后创建一个emr集群。

具体步骤

准备好mysql数据库环境，添加用户权限

首先需要一个mysql类型数据库（使用mysql协议，例如RDS、DRDS等),
数据库开启binlog，且配置binlog类型为ROW模式（RDS默认开启）。

  1. 先检查下环境：

  ### 查看是否开启binlog
  mysql> show variables like "log_bin";
  +---------------+-------+
  | Variable_name | Value |
  +---------------+-------+
  | log_bin       | ON    |
  +---------------+-------+
  1 row in set (0.02 sec)
  ### 查看binlog类型
  mysql> show variables like "binlog_format";
  +---------------+-------+
  | Variable_name | Value |
  +---------------+-------+
  | binlog_format | ROW   |
  +---------------+-------+
  1 row in set (0.03 sec)

  2. 添加权限，当然也可以直接通过rds控制台添加：

  CREATE USER canal IDENTIFIED BY 'canal';
  GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
  FLUSH PRIVILEGES;

开通sls，添加对应的配置文件，并检查数据是否正常采集

需要开通sls,并在sls控制台添加对应的project和logstore，例如：我创建的project叫canaltest，logstore叫canal.
然后对sls进行配置，配置文件目录在：/etc/ilogtail下创建文件:user_local_config.json，具体配置文件如下：

 {
      "metrics": {
          "##1.0##canaltest$plugin-local": {
              "aliuid": "****",
              "enable": true,
              "category": "canal",
              "defaultEndpoint": "*******",
              "project_name": "canaltest",
              "region": "cn-hangzhou",
              "version": 2
              "log_type": "plugin",
              "plugin": {
                  "inputs": [
                      {
                          "type": "service_canal",
                          "detail": {
                              "Host": "*****",
                              "Password": "****",
                              "ServerID": ****,
                              "User" : "***",
                              "DataBases": [
                                  "yourdb"
                              ],
                              "IgnoreTables": [
                                  "\\S+_inner"
                              ],
                               "TextToString" : true
                          }
                      }
                  ],
                  "flushers": [
                      {
                          "type": "flusher_sls",
                          "detail": {}
                      }
                  ]
              }
          }
      }
  }

其中detail中的Host和Password等信息为MySQL数据库信息，User请用之前授权过的用户名。
aliUid、defaultEndpoint、project_name、category请根据自己的实际情况填写对应的用户和sls信息。
配置好后，我们过2分钟左右，通过sls控制台，看下数据有没有上来，具体如图：

如果没有正常采集上来，请根据sls的提示，查看sls的采集日志进行排查。

准备好代码，编译成jar包，上传到oss

我们可以将emr的example代码通过git复制下来，修改下，具体命令如下：

git clone https://github.com/aliyun/aliyun-emapreduce-demo.git

example代码中已经有LoghubSample这个类，主要用于从sls采集数据，然后直接打印出来。
我们只要稍做修改，就可以达到我们的目的了。
以下是我修改后的代码：

package com.aliyun.emr.example

  import org.apache.spark.SparkConf
  import org.apache.spark.storage.StorageLevel
  import org.apache.spark.streaming.aliyun.logservice.LoghubUtils
  import org.apache.spark.streaming.{Milliseconds, StreamingContext}

  object LoghubSample {
    def main(args: Array[String]): Unit = {
      if (args.length < 7) {
        System.err.println(
          """Usage: bin/spark-submit --class LoghubSample examples-1.0-SNAPSHOT-shaded.jar
            |            
            |           
          """.stripMargin)
        System.exit(1)
      }

      val loghubProject = args(0)
      val logStore = args(1)
      val loghubGroupName = args(2)
      val endpoint = args(3)
      val accessKeyId = args(4)
      val accessKeySecret = args(5)
      val batchInterval = Milliseconds(args(6).toInt * 1000)

      val conf = new SparkConf().setAppName("Mysql Sync")
  //    conf.setMaster("local[4]");
      val ssc = new StreamingContext(conf, batchInterval)
      val loghubStream = LoghubUtils.createStream(
        ssc,
        loghubProject,
        logStore,
        loghubGroupName,
        endpoint,
        1,
        accessKeyId,
        accessKeySecret,
        StorageLevel.MEMORY_AND_DISK)

      loghubStream.foreachRDD(rdd =>
          rdd.saveAsTextFile("/mysqlbinlog")
      )

      ssc.start()
      ssc.awaitTermination()
    }
  }

其中主要改动是：

loghubStream.foreachRDD(rdd =>
       rdd.saveAsObjectFile("/mysqlbinlog")
)

这样在emr里运行时，就会把spark streaming中流出来的数据，保存到emr的hdfs中。

注意：

1. 由于如果要在本地运行起来，请在本地环境提前搭建hadoop集群。
2. 由于emr的spark sdk做了升级，他的example code有点旧，不能直接在参数中传递oss的accessKeyId、accessKeySecret 而是需要通过sparkConf设置进来。

trait RunLocally {
  val conf = new SparkConf().setAppName(getAppName).setMaster("local[4]")
  conf.set("spark.hadoop.fs.oss.impl", "com.aliyun.fs.oss.nat.NativeOssFileSystem")
  conf.set("spark.hadoop.mapreduce.job.run-local", "true")
  conf.set("spark.hadoop.fs.oss.endpoint", "YourEndpoint")
  conf.set("spark.hadoop.fs.oss.accessKeyId", "YourId")
  conf.set("spark.hadoop.fs.oss.accessKeySecret", "YourSecret")
  conf.set("spark.hadoop.job.runlocal", "true")
  conf.set("spark.hadoop.fs.oss.impl", "com.aliyun.fs.oss.nat.NativeOssFileSystem")
  conf.set("spark.hadoop.fs.oss.buffer.dirs", "/mnt/disk1")


  val sc = new SparkContext(conf)

  def getAppName: String
}

1. 在本地调试时，需要把


loghubStream.foreachRDD(rdd =>
rdd.saveAsObjectFile("/mysqlbinlog")
)

中的/mysqlbinlog修改成本地hdfs的地址

• 代码编译

在本地调试完成后，我们可以通过如下命令进行打包编译：

mvn clean install

• 上传jar包

通过oss控制台或oss的sdk将target/shaded目录下的examples-1.1-shaded.jar上传到oss上

注意，这个地址在后续会用上，如下图：

搭建emr集群，创建任务，运行执行计划，查看结果

1. 首先我们需要通过emr控制台创建一个emr集群，这个过程比较漫长，需要10分钟左右，请耐心等待
2. 创建类型为spark的作业
   

   --master yarn --deploy-mode client --driver-memory 4g --executor-memory 2g --executor-cores 2 --class com.aliyun.emr.example.LoghubSample ossref://emr-test/jar/examples-1.1-shaded.jar canaltest canal sparkstreaming $sls_endpoint $sls_access_id $sls_secret_key 1
   
   

   
   

请根据你的具体配置将

$sl{s}_{e}ndpoint$sls_access_id $sls_secret_key替换成真实值
注意参数的顺序，否则会报错。
3. 创建执行计划，将作业和emr集群绑定后，开始运行
4. 找到master节点的ip，如图：

通过ssh登录上去后，执行命令：

hadoop fs -ls /

hadoop fs -ls /mysqlbinlog

最后还可以通过命令：

hadoop fs -cat /mysqlbinlog/part-00000

查看文件内容，是不是很神奇呀?

错误排查

如果没有出来正常的结果，可以通过emr的运行记录，来进行问题排查,如图：