使用EMR进行MySQLBinlog日志准实时传输_开源大数据平台 E-MapReduce(EMR)-阿里云帮助中心

本文介绍如何利用阿里云SLS插件功能和E-MapReduce集群进行MySQL Binlog的准实时传输。

前提条件

已在E-MapReduce上创建Hadoop集群，详情请参见创建集群。
已创建MySQL类型的数据库（例如RDS或DRDS）。MySQL必须开启Binlog，且Binlog必须为ROW模式。
本文以RDS为例介绍，详情请参见创建RDS MySQL实例。
说明
RDS默认已开启Binlog功能。

操作步骤

连接MySQL实例并添加用户权限。
1. 使用命令方式连接MySQL实例，详情请参见通过客户端、命令行连接RDS。
2. 执行以下命令添加用户权限。
```
CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
```
为SLS服务添加对应的配置文件，详情请参见采集MySQL Binlog。
说明
本文创建的Project名称为canaltest，Logstore名称为canal。
在SLS控制台查看日志数据是否上传成功，如果未上传成功请根据SLS的采集日志排查。

编译JAR包并上传至OSS。

在本地打开Git复制示例代码。

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

修改示例代码。

示例代码中已经有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") )为loghubStream.foreachRDD(rdd => rdd.saveAsTextFile("/mysqlbinlog"))，以便于在E-MapReduce中运行时，保存Spark Streaming中流出来的数据至EMR的HDFS。

您可以在本地完成代码调试后，通过如下命令打包。
```
mvn clean install
```
上传JAR包至OSS。
在OSS上创建存储空间和上传文件，详情请参见控制台创建存储空间和控制台上传文件。
说明
本示例在OSS上创建的Bucket为EMR-test，上传examples-1.1-shaded.jar至EMR-test/jar目录。

提交并运行Spark作业。

spark-submit --master yarn --deploy-mode client \
  --driver-memory 4g --executor-memory 2g --executor-cores 2 \
  --class com.aliyun.EMR.example.LoghubSample \
  oss://EMR-test/jar/examples-1.1-shaded.jar \
  canaltest canal sparkstreaming cn-hangzhou-intranet.log.aliyuncs.com \
  $ALIBABA_CLOUD_ACCESS_KEY_ID $ALIBABA_CLOUD_ACCESS_KEY_SECRET 1

说明

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

执行如下命令查看mysqlbinlog目录下的文件。
```
hadoop fs -ls /mysqlbinlog
```
您还可以通过执行命令hadoop fs -cat /mysqlbinlog/part-00000查看文件内容。