婷婷久久综合九色综合97,无码国产福利片免费看,国产精品1区2区

本文介紹如何利用阿里云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查看文件內容。

日本熟妇hd丰满老熟妇,中文字幕一区二区三区在线不卡 ,亚洲成片在线观看,免费女同在线一区二区

使用E-MapReduce進行MySQL Binlog日志準實時傳輸

前提條件

操作步驟