@
目录

• FlinkX的安装与简单使用
  • FlinkX的安装
  • FlinkX的简单使用
    • 读取mysql中student表中数据
    • FlinkX本地运行
    • MySQLToHDFS
    • MySQLToHive
    • MySQLToHBase
    • MySQLToMySQL

FlinkX的安装与简单使用 @
目录

• FlinkX的安装与简单使用
  • FlinkX的安装
  • FlinkX的简单使用
    • 读取mysql中student表中数据
    • FlinkX本地运行
    • MySQLToHDFS
    • MySQLToHive
    • MySQLToHBase
    • MySQLToMySQL

FlinkX的安装

安装unzip：yum install unzip

【数据集成工具—FlinkX】1、上传并解压

unzip flinkx-1.10.zip -d /usr/local/soft/

2、配置环境变量
3、给bin/flinkx这个文件加上执行权限

chmod a+x flinkx

4、修改配置文件，设置运行端口

vim flinkconf/flink-conf.yaml

## web服务端口，不指定的话会随机生成一个 rest.bind-port: 8888

配置环境变量、
vim /etc/profile
FLINKX_HOME=

flinkX开源网址：https://github.com/DTStack/flinkx

FlinkX的简单使用
读取mysql中student表中数据

{ "job": { "content": [ { "reader": { "parameter": { "username": "root", "password": "123456", "connection": [{ "jdbcUrl": ["jdbc:mysql://master:3306/student?userSSL=false&useUnicode=true&characterEncoding=utf8"], "table": ["student"] }], "column": ["*"], "customSql": "", "where": "id > 1500100900", "splitPk": "id", "queryTimeOut": 1000 }, "name": "mysqlreader" }, "writer": { "name": "streamwriter", "parameter": { "print": true } } } ], "setting": { "speed": { "channel": 3, "bytes": 0 }, "errorLimit": { "record": 100 }, "restore": { "maxRowNumForCheckpoint": 0, "isRestore": false, "restoreColumnName": "", "restoreColumnIndex": 0 }, "log" : { "isLogger": false, "level" : "debug", "path" : "", "pattern":"" } } } }

FlinkX本地运行

flinkx -mode local -job flinkx3.json -pluginRoot ../syncplugins -flinkconf ../flinkconf 运行时文件所处路径为：/usr/local/soft/flinkx-1.10/package

MySQLToHDFS

• 配置文件

{ "job": { "content": [ { "reader": { "parameter": { "username": "root", "password": "123456", "connection": [ { "jdbcUrl": [ "jdbc:mysql://master:3306/student?characterEncoding=utf8" ], "table": [ "student" ] } ], "column": [ "*" ], "customSql": "", "where": "clazz = '理科二班'", "splitPk": "", "queryTimeOut": 1000, "requestAccumulatorInterval": 2 }, "name": "mysqlreader" }, "writer": { "name": "hdfswriter", "parameter": { "path": "hdfs://master:9000/data/flinkx/student", "defaultFS": "hdfs://master:9000", "column": [ { "name": "col1", "index": 0, "type": "string" }, { "name": "col2", "index": 1, "type": "string" }, { "name": "col3", "index": 2, "type": "string" }, { "name": "col4", "index": 3, "type": "string" }, { "name": "col5", "index": 4, "type": "string" }, { "name": "col6", "index": 5, "type": "string" } ], "fieldDelimiter": ",", "fileType": "text", "writeMode": "overwrite" } } } ], "setting": { "restore": { "isRestore": false, "isStream": false }, "errorLimit": {}, "speed": { "channel": 1 } } } }

• 启动任务

flinkx -mode local -job /usr/local/soft/flinkx-1.10/jsonConf/mysqlToHDFS.json -pluginRoot /usr/local/soft/flinkx-1.10/syncplugins/ -flinkconf /usr/local/soft/flinkx-1.10/flinkconf/

• 监听日志

flinkx 任务启动后，会在执行命令的目录下生成一个nohup.out文件

tail -f nohup.out

• 通过web界面查看任务运行情况

http://master:8888

MySQLToHive

• 配置文件

{ "job": { "content": [ { "reader": { "parameter": { "username": "root", "password": "123456", "connection": [ { "jdbcUrl": [ "jdbc:mysql://master:3306/student?characterEncoding=utf8" ], "table": [ "student" ] } ], "column": [ "*" ], "customSql": "", "where": "clazz = '文科二班'", "splitPk": "id", "queryTimeOut": 1000, "requestAccumulatorInterval": 2 }, "name": "mysqlreader" }, "writer": { "name": "hivewriter", "parameter": { "jdbcUrl": "jdbc:hive2://master:10000/testflinkx", "username": "", "password": "", "fileType": "text", "fieldDelimiter": ",", "writeMode": "overwrite", "compress": "", "charsetName": "UTF-8", "maxFileSize": 1073741824, "tablesColumn": "{\"student\":[{\"key\":\"id\",\"type\":\"string\"},{\"key\":\"name\",\"type\":\"string\"},{\"key\":\"age\",\"type\":\"string\"}]}", "defaultFS": "hdfs://master:9000" } } } ], "setting": { "restore": { "isRestore": false, "isStream": false }, "errorLimit": {}, "speed": { "channel": 3 } } } }

• 在hive中创建testflinkx数据库，并创建student分区表

create database testflinkx; use testflinkx; CREATE TABLE `student`( `id` string, `name` string, `age` string) PARTITIONED BY ( `pt` string) ROW FORMAT DELIMITED FIELDS TERMINATED BY ','

• 启动hiveserver2

# 第一种方式： hiveserver2 # 第二种方式: hive --service hiveserver2

• 启动任务

flinkx -mode local -job /usr/local/soft/flinkx-1.10/jsonConf/mysqlToHive.json -pluginRoot /usr/local/soft/flinkx-1.10/syncplugins/ -flinkconf /usr/local/soft/flinkx-1.10/flinkconf/

• 查看日志及运行情况同上

MySQLToHBase

• 配置文件

{ "job": { "content": [ { "reader": { "parameter": { "username": "root", "password": "123456", "connection": [ { "jdbcUrl": [ "jdbc:mysql://master:3306/student?characterEncoding=utf8" ], "table": [ "score" ] } ], "column": [ "*" ], "customSql": "", "splitPk": "student_id", "queryTimeOut": 1000, "requestAccumulatorInterval": 2 }, "name": "mysqlreader" }, "writer": { "name": "hbasewriter", "parameter": { "hbaseConfig": { "hbase.zookeeper.property.clientPort": "2181", "hbase.rootdir": "hdfs://master:9000/hbase", "hbase.cluster.distributed": "true", "hbase.zookeeper.quorum": "master,node1,node2", "zookeeper.znode.parent": "/hbase" }, "table": "testFlinkx", "rowkeyColumn": "$(cf1:student_id)_$(cf1:course_id)", "column": [ { "name": "cf1:student_id", "type": "string" }, { "name": "cf1:course_id", "type": "string" }, { "name": "cf1:score", "type": "string" } ] } } } ], "setting": { "restore": { "isRestore": false, "isStream": false }, "errorLimit": {}, "speed": { "channel": 3 } } } }

• 启动hbase 并创建testflinkx表

create 'testFlinkx','cf1'

• 启动任务

flinkx -mode local -job /usr/local/soft/flinkx-1.10/jsonConf/mysqlToHBase.json -pluginRoot /usr/local/soft/flinkx-1.10/syncplugins/ -flinkconf /usr/local/soft/flinkx-1.10/flinkconf/

• 查看日志及运行情况同上

MySQLToMySQL

• 配置文件

{ "job": { "content": [ { "reader": { "name": "mysqlreader", "parameter": { "column": [ { "name": "id", "type": "int" }, { "name": "name", "type": "string" }, { "name": "age", "type": "int" }, { "name": "gender", "type": "string" }, { "name": "clazz", "type": "string" } ], "username": "root", "password": "123456", "connection": [ { "jdbcUrl": [ "jdbc:mysql://master:3306/student?useSSL=false" ], "table": [ "student" ] } ] } }, "writer": { "name": "mysqlwriter", "parameter": { "username": "root", "password": "123456", "connection": [ { "jdbcUrl": "jdbc:mysql://master:3306/student?useSSL=false", "table": [ "student2" ] } ], "writeMode": "insert", "column": [ { "name": "id", "type": "int" }, { "name": "name", "type": "string" }, { "name": "age", "type": "int" }, { "name": "gender", "type": "string" }, { "name": "clazz", "type": "string" } ] } } } ], "setting": { "speed": { "channel": 1, "bytes": 0 } } } }

推荐阅读

• 

  信号工证怎么查询 信号工证查询步骤
• 

  佳能ts208打印机清零
• 

  抽血后容易疲劳是怎么回事
• 

  济南摄影活动 济南延时摄影
• 

  你了解Win7文件夹高级权限设置吗
• 

  鲤鱼汤的做法 鲤鱼汤的做法简介
• 

  纵观国内娱乐圈,你觉得需要整顿吗？
• 

  橙子会让皮肤变黄吗
• 

  送给好朋友短信生日祝福语
• 

  冻梨可以用热水泡吗
• 

  学考过不了能上大学吗
• 

  荷兰豆|“荷兰豆”并非产自荷兰
• 

  你好，宝宝二十天了小儿脐风散误食了一袋的量，因该怎么处理……谢谢
• 

  第五人格印记怎么开启 第五人格印记怎么开启不了
• 

  如何让宫口开得快
• 

  手机电池里的液体弄手上了？有危险吗？
• 

  脑梗|半年被脑梗“偷袭”两次，和早晨常做几件事，脱不了干系
• 

  公务员职位中法检系统中招录的法律类一级科员是什么？和四级检察官助理,法官助理有什么区别？
• 

  猎豹浏览器怎么设置兼容模式？兼容模式设置方法详解
• 

  郫县豆瓣史话

• Docker应用:容器间通信与Mariadb数据库主从复制
• Activiti（一）SpringBoot2集成Activiti6
• 标签、语法规范、内联框架、超链接、CSS的编写位置、CSS语法、开发工具、块和内联、常用选择器、后代元素选择器、伪类、伪元素。
• 孩子不是实现父母欲望的工具——林哈夫
• 使用协程爬取网页，计算网页数据大小
• 私有化轻量级持续集成部署方案--03-部署web服务（下）
• 最有效的时间管理工具（赢效率手册和总结笔记）
• 工具|后天就是七夕节，你准备好了吗（送上几个七夕代码，展示你技能的时候到了！）
• Java|Java基础——数组
• Python数据分析(一)（Matplotlib使用）