FlinkX学习

FlinkX学习

FlinkX安装

由于flinkx已经改名chunjun 官网已不存在

(https://gitee.com/lugela/flinkx#flinkx)这里可以看到flinkx的操作文档

1、上传并解压

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

2、配置环境变量

FLINKX_HOME=/usr/local/soft/flinkx-1.10
export PATH=$FLINKX_HOME/bin:$PATH

3、给bin/flinkx这个文件加上执行权限

 chmod +x flinkx

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

vim flinkconf/flink-conf.yaml## web服务端口,不指定的话会随机生成一个
rest.bind-port: 8888

启动

命令行参数选项

  • model
    • 描述:执行模式,也就是flink集群的工作模式
      • local: 本地模式
      • standalone: 独立部署模式的flink集群
      • yarn: yarn模式的flink集群,需要提前在yarn上启动一个flink session,使用默认名称"Flink session cluster"
      • yarnPer: yarn模式的flink集群,单独为当前任务启动一个flink session,使用默认名称"Flink per-job cluster"
    • 必选:否
    • 默认值:local
  • job
    • 描述:数据同步任务描述文件的存放路径;该描述文件中使用json字符串存放任务信息。
    • 必选:是
    • 默认值:无
  • pluginRoot
    • 描述:插件根目录地址,也就是打包后产生的pluginRoot目录。
    • 必选:是
    • 默认值:无
  • flinkconf
    • 描述:flink配置文件所在的目录(单机模式下不需要),如/opt/dtstack/flink-1.8.1/conf
    • 必选:否
    • 默认值:无
  • yarnconf
    • 描述:Hadoop配置文件(包括hdfs和yarn)所在的目录(单机模式下不需要),如/hadoop/etc/hadoop
    • 必选:否
    • 默认值:无
  • flinkLibJar
    • 描述:flink lib所在的目录(单机模式下不需要),如/opt/dtstack/flink-1.8.1/lib
    • 必选:否
    • 默认值:无
  • confProp
    • 描述:flink相关参数,如{“flink.checkpoint.interval”:200000}
    • 必选:否
    • 默认值:无
  • queue
    • 描述:yarn队列,如default
    • 必选:否
    • 默认值:无
  • pluginLoadMode
    • 描述:yarnPer模式插件加载方式:
      • classpath:提交任务时不上传插件包,需要在yarn-node节点pluginRoot目录下部署插件包,但任务启动速度较快
      • shipfile:提交任务时上传pluginRoot目录下部署插件包的插件包,yarn-node节点不需要部署插件包,任务启动速度取决于插件包的大小及网络环境
    • 必选:否
    • 默认值:classpath

FlinkX概述

FlinkX是在是袋鼠云内部广泛使用的基于flink的分布式离线和实时的数据同步框架,实现了多种异构数据源之间高效的数据迁移。

不同的数据源头被抽象成不同的Reader插件,不同的数据目标被抽象成不同的Writer插件。理论上,FlinkX框架可以支持任意数据源类型的数据同步工作。作为一套生态系统,每接入一套新数据源该新加入的数据源即可实现和现有的数据源互通。

FlinkX是一个基于Flink的批流统一体的数据同步工具,既可以采集静态的数据,比如MySQL,HDFS等,也可以采集实时变化的数据,比如MySQL binlog,Kafka等

在底层实现上,FlinkX依赖Flink,数据同步任务会被翻译成StreamGraph在Flink上执行.

image-20220619223533456.png

FlinkX的简单使用

MySQL2HDFS

场景

将mysql Y1数据库下的Student表数据写入HDFS上的指定路径中

参考文档

mysqlreader:(https://gitee.com/lugela/flinkx/blob/1.10_release/docs/offline/reader/mysqlreader.md)

hdfswriter:(https://gitee.com/lugela/flinkx/blob/1.10_release/docs/offline/writer/hdfswriter.md)

创建mysql2hdfs.json文件

{"job": {"content": [{"reader": {"parameter": {"username": "root","password": "123456","connection": [{"jdbcUrl": ["jdbc:mysql://master:3306/Y1?useUnicode=true&characterEncoding=utf8&useSSL=false"],"table": ["Student"]}],"column": ["*"],"where": "Sid > 05 ","requestAccumulatorInterval": 2},"name": "mysqlreader"},"writer": {"name": "hdfswriter","parameter": {"path": "hdfs://master:9000/bigdata30/flinkx/out1","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"}],"fieldDelimiter": ",","fileType": "text","writeMode": "append"}}}],"setting": {"restore": {"isRestore": false,"isStream": false},"errorLimit": {},"speed": {"channel": 1}}}
}

运行模式

  • 单机模式:对应Flink集群的单机模式
  • standalone模式:对应Flink集群的分布式模式
  • yarn模式:对应Flink集群的yarn模式
  • yarnPer模式: 对应Flink集群的Per-job模式

运行:

flinkx -mode local -job ./mysql2hdfs.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

hdfs上出现文件:

image.png

查看该文件:

hdfs dfs -cat /bigdata30/flinkx/out1/0.44b7d6c8dcaadcc14ae55fb482f9fb27.0

出现Sid大于05的学生:

image.png

MySQLToHive

hivewrite:(https://github.com/oceanos/flinkx/blob/1.8_release/docs/hivewriter.md)

配置文件:

{"job": {"content": [{"reader": {"parameter": {"username": "root","password": "123456","connection": [{"jdbcUrl": ["jdbc:mysql://master:3306/Y1?useUnicode=true&characterEncoding=utf8&useSSL=false"],"table": ["Student"]}],"column": ["*"],"where": "Sid > 05 ","requestAccumulatorInterval": 2},"name": "mysqlreader"},"writer": {"name": "hivewriter","parameter": {"jdbcUrl": "jdbc:hive2://master:10000/bigdata30","username": "","password": "","fileType": "text","fieldDelimiter": ",","writeMode": "overwrite","compress": "","charsetName": "UTF-8","maxFileSize": 1073741824,"tablesColumn": "{\"Student\":[{\"key\":\"SId\",\"type\":\"string\"},{\"key\":\"Sname\",\"type\":\"string\"},{\"key\":\"Sage\",\"type\":\"string\"},{\"key\":\"Ssex\",\"type\":\"string\"}]}","defaultFS": "hdfs://master:9000"}}}],"setting": {"restore": {"isRestore": false,"isStream": false},"errorLimit": {},"speed": {"channel": 1}}}
}

在hive中建表:

CREATE TABLE `bigdata30`.`Student`(`SId` STRING,`Sname` STRING,`Sage` STRING,`Ssex` STRING)
PARTITIONED BY ( `pt` string)
ROW FORMAT DELIMITED FIELDS TERMINATED BY ',';

启动hiveserver2

启动任务

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

运行发现报错 无法解决。

翻阅chunjun官网 在hive-sink中发现 只支持hive1.x hive2.x 现hive版本为3.1.2 不支持 猜测报错原因

尝试使用chunjun 解决

MySQLToHBase

场景

将mysql Y1数据库中的Student表数据写入HBase flinkx_Student表中

配置文件

{"job": {"content": [{"reader": {"parameter": {"username": "root","password": "123456","connection": [{"jdbcUrl": ["jdbc:mysql://master:3306/Y1?useUnicode=true&characterEncoding=utf8&useSSL=false"],"table": ["Student"]}],"column": ["*"],"where": "Sid > 05 ","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": "flinkx_Student","rowkeyColumn": "$(cf1:SId)","column": [{"name": "cf1:SId","type": "string"},{"name": "cf1:Sname","type": "string"},{"name": "cf1:Sage","type": "string"},{"name": "cf1:Ssex","type": "string"}]}}}],"setting": {"restore": {"isRestore": false,"isStream": false},"errorLimit": {},"speed": {"channel": 1}}}
}

在hbase中创建flinkx_Student表

create 'flinkx_Student','cf1'

启动

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

hbase中的flinkx_Student表出现数据

image.png

MySQLToMySQL

场景

将mysql Y1数据库中的Student表数据写入datax1数据库中的Student2表中

配置文件 mysql2mysql.json

{"job": {"content": [{"reader": {"name": "mysqlreader","parameter": {"column": [{"name": "SId","type": "string"},{"name": "Sname","type": "string"},{"name": "Sage","type": "string"},{"name": "Ssex","type": "string"}],"username": "root","password": "123456","connection": [{"jdbcUrl": ["jdbc:mysql://master:3306/Y1?useSSL=false"],"table": ["Student"]}]}},"writer": {"name": "mysqlwriter","parameter": {"username": "root","password": "123456","connection": [{"jdbcUrl": "jdbc:mysql://master:3306/datax1?useSSL=false","table": ["Student2"]}],"writeMode": "insert","column": [{"name": "SId","type": "string"},{"name": "Sname","type": "string"},{"name": "Sage","type": "string"},{"name": "Ssex","type": "string"}]}}}],"setting": {"speed": {"channel": 1,"bytes": 0}}}}

在mysql datax1数据库中建表:

create table if not exists datax1.Student2(SID varchar(10),Sname varchar(100),Sage varchar(100),Ssex varchar(10)
)CHARSET = utf8 COLLATE utf8_general_ci;

运行:

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

进入网页查看:

master:8888

image.png

查看Student2表 数据已导入:

image.png

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/bicheng/36534.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

STL-迭代器

1.迭代器 1.1正向迭代器 正向迭代器是用一个类封装的,迭代器类。例如:在vector,string中的迭代器就相当于一个指针,在list类中用一个类来封装一个节点,实质上也还是一个指针,迭代器就相当于指向一个节点的…

背靠广汽、小马智行,如祺出行打得过滴滴和百度吗?

©自象限原创 作者丨艾AA 编辑丨薛黎 北京时间6月14日凌晨,在特斯拉股东大会上,马斯克阐述了对Robotaxi(自动驾驶出租车)商业模式的构想——特斯拉不仅会运营自己的无人驾驶出租车车队,还可以让特斯拉车主们的爱…

微服务开发 —— 项目环境搭建篇

环境搭建 Linux 环境搭建 Linux 环境搭建大家可以使用虚拟机 VMware、VirtualBox 等应用创建虚拟机,使用Vagrant也可以快捷搭建虚拟环境;Windows 中有 WSL2,Windows 中的 Docker 也对 WSL 进行了支持,也是一个不错的选择。或者可…

Windows - 像Linux一样使用alias

前言 阅读本文约2分钟 说明 使用doskey命令 设置别名 查看当前目录 doskey lldir 激活conda的spider环境 doskey spiderconda activate spider 退出conda环境 doskey condaqconda deactivate 查看所有别名 doskey /macros

实现设计开挂|如何设计出坚不可摧的网球拍?

数字揭秘 我们发现自己可能偶尔会以过激的方式表达沮丧或愤怒,哪怕是在公共场合。就算是世界级的网球运动员也无法避免偶尔的情绪爆发,他们有时会砸球拍来释放被压抑的情绪或应对来自竞赛的压力。 网球运动员的情绪爆发已被证明是不可避免的。哪怕是包括…

Linux系统相关函数总结

在应用程序当中,有时往往需要去获取到一些系统相关的信息,譬如时间、日期、以及其它一些系统相关信息,本章将向大家介绍如何通过 Linux 系统调用或 C 库函数获取这些系统信息。除此之外,还会向大家介绍 Linux 系统下的/proc 虚拟文…

浏览器插件利器-allWebPluginV2.0.0.14-beta版发布

allWebPlugin简介 allWebPlugin中间件是一款为用户提供安全、可靠、便捷的浏览器插件服务的中间件产品,致力于将浏览器插件重新应用到所有浏览器。它将现有ActiveX插件直接嵌入浏览器,实现插件加载、界面显示、接口调用、事件回调等。支持谷歌、火狐等浏…

js删除el-table删除新增项,有的已经保存有的未经保存

有时候在弹窗中的弹窗要删除数据,有的是刚新增进来的,没有经过保存就没有id,有的已经保存过就有id 根据情况设定是否为编辑模式,如果为编辑模式就需要进行筛选删除及接口,如果不是编辑模式,只需要进行筛选删除 this.editFlag true; // 为编辑模式// 删除伤亡名单handelDel() …

数据库管理系统(DBMS)

一.数据库管理系统 1.简介 数据库管理系统(Database Management System)是一种操纵和管理数据库的大型软件,用于建立、使用和维护数据库,简称DBMS。它对数据库进行统一的管理和控制,以保证数据库的安全性和完整性。用户通过DBMS访问数据库中…

【PWN · ret2libc | protobuf】[2024CISCN · 华中赛区]protoverflow

套了一层protobuf壳,然后就是简单的ret2libc 参考速递:深入二进制安全:全面解析Protobuf-CSDN博客 前言 第一次遇到protobuf,如果没有了解过,是显然做不出来的。此次复现,也算是点亮了一个技能点 一、什么…

如何用Vue3和Plotly.js实现一个动态3D图的在线展示

本文由ScriptEcho平台提供技术支持 项目地址:传送门 基于 Plotly.js 的交互式图表动画 应用场景 本代码演示了如何使用 Plotly.js 创建交互式图表动画,其中一个区域填充的区域在给定时间间隔内更新其数据。这种动画可用于可视化时间序列数据或展示数…

centOS7网络配置_NAT模式设置

第一步:查看电脑网卡 nat模式对应本地网卡的VMnet 8 ,查看对应的IP地址。 第二步:虚拟网络编辑器 打开VMWare,编辑--虚拟网络编辑器,整个都默认设置好了,只需要查看对应的DHCP设置中对应的IP的起始&#…

算法金 | 没有思考过 Embedding,不足以谈 AI

大侠幸会,在下全网同名「算法金」 0 基础转 AI 上岸,多个算法赛 Top 「日更万日,让更多人享受智能乐趣」 抱个拳,送个礼 在当今的人工智能(AI)领域,Embedding 是一个不可或缺的概念。如果你没有…

Python学习笔记24:进阶篇(十三)常见标准库使用之数据压缩功能模块zlib,gzip,bz2,lzma的学习使用

前言 本文是根据python官方教程中标准库模块的介绍,自己查询资料并整理,编写代码示例做出的学习笔记。 根据模块知识,一次讲解单个或者多个模块的内容。 教程链接:https://docs.python.org/zh-cn/3/tutorial/index.html 数据压缩…

2024年度临沂市安全文化书画摄影展开幕

人海信息网山东讯 6月27日,2024年度临沂市安全文化书画摄影作品展,在临沂高新区隆重开幕。本次书画摄影展深入贯彻“以人为本,安全发展”的重要思想,立意高远,内涵丰富,思想深邃,承载着健康、幸…

List常用操作比for循环更优雅的写法

private String name; //姓名 private Integer age; //年龄 private Integer departId; //所属部门id } List list new ArrayList<>(); 复制代码 简单遍历 使用lamada表达式之前&#xff0c;如果需要遍历list时&#xff0c;一般使用增强for循环&#xff0c;代码如…

【AIGC】《AI-Generated Content (AIGC): A Survey》

文章目录 相关概念What is AI-generated content?Necessary conditions of AIGCHow can AI make the content better?The industrial chain of AIGCAdvantages of large-scale pre-trained modelsGeneration of smart textPros of AIGCCons of AIGCAIGC and Metaverse 挑战潜…

制作高校专属PPT时,如何将校徽设置成透明底色?无须PS

目录 示例&#xff1a;以清华大学为例 1必应搜索“清华大学校徽” 2保存清华大学校徽及校名。 3将校徽导入到PPT中 ​4 选中校徽&#xff0c;然后依次选择“图片格式”-->颜色-->设置透明色​编辑 5出现“画笔”&#xff0c;由于截图的缘故&#xff0c;画笔没有在截…

002关于Geogebra软件的介绍及与MatLab的区别

为什么要学Geogebra&#xff1f; 因为和MatLab的科学计算相比&#xff0c;GeoGebra重点突出教学展示&#xff0c;对于教师、学生人群来讲再合适不过了&#xff0c;尤其是可以融入到PPT里边呈现交互式动画&#xff0c;想想听众的表情&#xff01;这不就弥补了看到PPT播放数学公…

谷歌SEO在外贸推广中的应用效果如何?

谷歌SEO在外贸推广中非常有效。通过优化网站&#xff0c;可以提高在搜索结果中的排名&#xff0c;这意味着更多的潜在客户会看到你的产品和服务。 一个高排名的网站能带来更多自然流量&#xff0c;不需要花费广告费用。这种流量通常质量较高&#xff0c;因为用户是主动搜索相关…