全部产品
Search
文档中心

开源大数据平台E-MapReduce:使用E-MapReduce进行MySQL Binlog日志准实时传输

更新时间:Sep 14, 2023

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

前提条件

  • 已在E-MapReduce上创建Hadoop集群,详情请参见创建集群

  • 已创建MySQL类型的数据库(例如RDS或DRDS)。MySQL必须开启Binlog,且Binlog必须为ROW模式。

    本文以RDS为例介绍,详情请参见创建RDS MySQL实例

    说明

    RDS默认已开启Binlog功能。

操作步骤

  1. 连接MySQL实例并添加用户权限。

    1. 使用命令方式连接MySQL实例,详情请参见通过客户端、命令行连接RDS

    2. 执行以下命令添加用户权限。

      CREATE USER canal IDENTIFIED BY 'canal';
      GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
      FLUSH PRIVILEGES;
  2. 为SLS服务添加对应的配置文件,详情请参见采集MySQL Binlog

    说明

    本文创建的Project名称为canaltest,Logstore名称为canal。

    在SLS控制台查看日志数据是否上传成功,如果未上传成功请根据SLS的采集日志排查。

  3. 编译JAR包并上传至OSS。

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

      git clone https://github.com/aliyun/aliyun-emapreduce-demo.git
    2. 修改示例代码。

      示例代码中已经有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。

    3. 您可以在本地完成代码调试后,通过如下命令打包。

      mvn clean install
    4. 上传JAR包至OSS。

      在OSS上创建存储空间和上传文件,详情请参见控制台创建存储空间控制台上传文件

      说明

      本示例在OSS上创建的Bucket为EMR-test,上传examples-1.1-shaded.jarEMR-test/jar目录。

  4. 提交并运行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
    说明

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

  5. 执行如下命令查看mysqlbinlog目录下的文件。

    hadoop fs -ls /mysqlbinlog

    您还可以通过执行命令hadoop fs -cat /mysqlbinlog/part-00000查看文件内容。