Mysql增量寫入Hdfs(一) --將Mysql數據寫入Kafka Topic


一. 概述

在大數據的靜態數據處理中,目前普遍采用的是用Spark+Hdfs(Hive/Hbase)的技術架構來對數據進行處理。

但有時候有其他的需求,需要從其他不同數據源不間斷得采集數據,然后存儲到Hdfs中進行處理。而追加(append)這種操作在Hdfs里面明顯是比較麻煩的一件事。所幸有了Storm這么個流數據處理這樣的東西問世,可以幫我們解決這些問題。

不過光有Storm還不夠,我們還需要其他中間件來協助我們,讓所有其他數據源都歸於一個通道。這樣就能實現不同數據源以及Hhdfs之間的解耦。而這個中間件Kafka無疑是一個很好的選擇。

這樣我們就可以讓Mysql的增量數據不停得拋出到Kafka,而后再讓storm不停得從Kafka對應的Topic讀取數據並寫入到Hdfs中。

二.binlog和maxwell介紹

2.1Mysql binlog介紹

binlog即Mysql的二進制日志。它可以說是Mysql最重要的日志了,它記錄了所有的DDL和DML(除了數據查詢語句)語句,以事件形式記錄,還包含語句所執行的消耗的時間,MySQL的二進制日志是事務安全型的。

上面所說的提到了DDL和DML,可能有些同學不了解,這里順便說一下:

  • DDL:數據定義語言DDL用來創建數據庫中的各種對象-----表、視圖、索引、同義詞、聚簇等如:CREATETABLE/VIEW/INDEX/SYN/CLUSTER...
  • DML:數據操縱語言DML主要有三種形式:插入(INSERT),更新(UPDATE),以及刪除(DELETE)。

在Mysql中,binlog默認是不開啟的,因為有大約1%(官方說法)的性能損耗,如果要手動開啟,流程如下:

  1. vi編輯打開mysql配置文件:
vi /usr/local/mysql/etc/my.cnf

在[mysqld]區塊設置/添加如下,

log-bin=mysql-bin 

注意一定要在[mysqld]下。
2. 重啟Mysql

pkill mysqld
/usr/local/mysql/bin/mysqld_safe --user=mysql &

2.2kafka

這里只對Kafka做一個基本的介紹,更多的內容可以度娘一波。

上面的圖片是kafka官方的一個圖片,我們目前只需要關注Producers和Consumers就行了。

Kafka是一個分布式發布-訂閱消息系統。分布式方面由Zookeeper進行協同處理。消息訂閱其實說白了吧,就是一個隊列,分為消費者和生產者,就像上圖中的內容,有數據源充當Producer生產數據到kafka中,而有數據充當Consumers,消費kafka中的數據。

上圖中的offset指的是數據的寫入以及消費的位置的信息,這是由Zookeeper管理的。也就是說,當Consumers重啟或是怎樣,需要重新從kafka讀取消息時,總不能讓它從頭開始消費數據吧,這時候就需要有個記錄能告訴你從哪里開始重新讀取。這就是offset。

kafka中還有一個至關重要的概念,那就是topic。不過這個其實還是很好理解的,比如你要訂閱一些消息,你肯定是不會訂閱所有消息的吧,你只需要訂閱你感興趣的主題,比如攝影,編程,搞笑這些主題。而這里主題的概念其實和topic是一樣的。總之,可以將topic歸結為通道,kafka中有很多個通道,不同的Producer向其中一個通道生產數據,也就是拋數據進去這個通道,Comsumers不停得消費通道中的數據。

而我們要做的就是將Mysql binlog產生的數據拋到kafka中充當作生產者,然后由storm充當消費者,不停得消費數據並寫入到Hdfs中。

至於怎么將binlog的數據拋到kafka,別急,下面我們就來介紹。

2.3maxwell

maxwell這個工具可以很方便得監聽Mysql的binlog,然后每當binlog發生變化時,就會以json格式拋出對應的變化數據到Kafka中。比如當向mysql一張表中插入一條語句的時候,maxwell就會立刻監聽到binlog中有對應的記錄增加,然后將一些信息包括插入的數據都轉化成json格式,然后拋到kafka指定的topic中。

下載地址在這里可以找到。

除了Kafka外,其實maxwell還支持寫入到其他各種中間件,比如redis。
同時maxwell是比較輕量級的工具,只需要在mysql中新建一個數據庫供它記錄一些信息,然后就可以直接運行。

三.使用maxwell監聽binlog

接下來我們將的是如果使用maxwell,讓它監聽mysql的binlog並拋到kafka中。maxwell主要有兩種運行方式。一種是使用配置文件,另一種則是在命令行中添加參數的方式運行。這里追求方便,只使用命令行的方式進行演示。

這里介紹一下簡單的將數據拋到kafka的命令行腳本吧。

bin/maxwell --user='maxwell' --password='XXXXXX' --host='127.0.0.1' \
   --producer=kafka --kafka.bootstrap.servers=localhost:9092 --kafka_topic=maxwell --port=3306

各項參數說明如下

  • user:mysql用戶名
  • password:mysql密碼
  • host:Mysql地址
  • producer:指定寫入的中間件類型,比如還有redies
  • kafka.bootstrap.servers:kafka的地址
  • kafka_topic:指明寫入到kafka哪個topic
  • port:mysql端口

啟動之后,maxwell便開始工作了,當然如果你想要讓這條命令可以在后台運行的話,可以使用Linux的nohup命令,這里就不多贅述,有需要百度即可。

這樣配置的話通常會將整個數據庫的增刪改都給拋到kafka,但這樣的需求顯然不常見,更常見的應該是具體監聽對某個庫的操作,或是某個表的操作。

在升級到1.9.2(最新版本)后,maxwell為我們提供這樣一個參數,讓我們可以輕松實現上述需求:--filter

這個參數通常包含兩個配置項,exclude和include。意思就是讓你指定排除哪些和包含哪些。比如我只想監聽Adatabase庫下的Atable表的變化。我可以這樣。

--filter='exclude: *.*, include: Adatabase.Atable'

這樣我們就可以輕松實現監聽mysqlbinlog的變化,並可以定制自己的需求。

OK,這一章我們介紹了mysql binlog,kafka以及maxwell的一些內容,下一篇我們將會看到storm如何寫入hdfs以及定制一些策略。see you~~

推薦閱讀 :
從分治算法到 MapReduce
一個故事告訴你什么才是好的程序員
大數據存儲的進化史 --從 RAID 到 Hadoop Hdfs


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM