- Java锁的逻辑(结合对象头和ObjectMonitor)
- 还在用饼状图?来瞧瞧这些炫酷的百分比可视化新图形(附代码实现)⛵
- 自动注册实体类到EntityFrameworkCore上下文,并适配ABP及ABPVNext
- 基于Sklearn机器学习代码实战
@ 。
Chunjun 官网 https://dtstack.github.io/chunjun-web/ 源码release最新版本1.12.8 。
Chunjun 文档地址 https://ververica.github.io/flink-cdc-connectors/master/ 。
Chunjun 源码地址 https://github.com/DTStack/chunjun 。
Chunjun是一个分布式集成框架,原名是FlinkX,由袋鼠云开源,其基于Flink的批流统一打造的数据同步工具,可以实现各种异构数据源之间的数据同步和计算.
ChunJun是一个基于 Flink 提供易用、稳定、高效的批流统一的数据集成工具,可以采集静态的数据如 MySQL,HDFS 等,也可以采集实时变化的数据如 binlog,Kafka等.
# 最新release版本源码flink12.7,如果是下载主线master版本,目前源码默认引入flink16.1,可以通过git clone https://github.com/DTStack/chunjun.git也可以直接http下main,由于是学习可使用master版本来踩坑
wget https://github.com/DTStack/chunjun/archive/refs/tags/v1.12.8.tar.gz
tar -xvf v1.12.8.tar.gz
# 进入源码目录
cd chunjun-1.12.8/
# 编译打包执行,下面两种选一
./mvnw clean package
sh build/build.sh
在根目录下生成 chunjun-dist 目录,官方提供丰富的示例程序,详细可以查看chunjun-examples目录 。
下表显示了ChunJun分支与flink版本的对应关系。如果版本没有对齐,在任务中会出现'Serialization Exceptions', 'NoSuchMethod Exception'等问题.
一个完整的 ChunJun 任务脚本配置包含 content, setting 两个部分。content 用于配置任务的输入源与输出源,其中包含 reader,writer。而 setting 则配置任务整体的环境设定,其中包含 speed,errorLimit,metricPluginConf,restore,log,dirty。总体结构如下所示:
{
"job" : {
"content" :[{
"reader" : {},
"writer" : {}
}],
"setting" : {
"speed" : {},
"errorLimit" : {},
"metricPluginConf" : {},
"restore" : {},
"log" : {},
"dirty":{}
}
}
}
reader 用于配置数据的输入源,即数据从何而来。具体配置如下所示:
"reader" : {
"name" : "xxreader",
"parameter" : {
......
}
}
Writer 用于配置数据的输出目的地,即数据写到哪里去。具体配置如下所示:
"writer" : {
"name" : "xxwriter",
"parameter" : {
......
}
}
详细使用查看官方的说明 。
进入Chunjun根目录,测试脚本执行本地环境,查看stream.json 。
{
"job": {
"content": [
{
"reader": {
"parameter": {
"column": [
{
"name": "id",
"type": "id"
},
{
"name": "name",
"type": "string"
},
{
"name": "content",
"type": "string"
}
],
"sliceRecordCount": [
"30"
],
"permitsPerSecond": 1
},
"table": {
"tableName": "sourceTable"
},
"name": "streamreader"
},
"writer": {
"parameter": {
"column": [
{
"name": "id",
"type": "id"
},
{
"name": "name",
"type": "string"
}
],
"print": true
},
"table": {
"tableName": "sinkTable"
},
"name": "streamwriter"
},
"transformer": {
"transformSql": "select id,name from sourceTable where CHAR_LENGTH(name) < 50 and CHAR_LENGTH(content) < 50"
}
}
],
"setting": {
"errorLimit": {
"record": 100
},
"speed": {
"bytes": 0,
"channel": 1,
"readerChannel": 1,
"writerChannel": 1
}
}
}
}
bash ./bin/chunjun-local.sh -job chunjun-examples/json/stream/stream.json
将依赖文件复制到Flink lib目录下,这个复制操作需要在所有Flink cluster机器上执行 。
cp -r chunjun-dist $FLINK_HOME/lib
启动Flink Standalone环境 。
sh $FLINK_HOME/bin/start-cluster.sh
准备mysql的数据,作为读取数据源 。
准备job文件,创建chunjun-examples/json/mysql/mysql_hdfs_polling_my.json 。
{
"job": {
"content": [
{
"reader": {
"name": "mysqlreader",
"parameter": {
"column" : [
{
"name" : "id",
"type" : "bigint"
},{
"name" : "name",
"type" : "varchar"
},{
"name" : "age",
"type" : "bigint"
}
],
"splitPk": "id",
"splitStrategy": "mod",
"increColumn": "id",
"startLocation": "1",
"username": "root",
"password": "123456",
"queryTimeOut": 2000,
"connection": [
{
"jdbcUrl": [
"jdbc:mysql://mysqlserver:3308/my_maxwell_01?useSSL=false"
],
"table": [
"account"
]
}
],
"polling": false,
"pollingInterval": 3000
}
},
"writer": {
"name": "hdfswriter",
"parameter": {
"fileType": "text",
"path": "hdfs://myns/user/hive/warehouse/chunjun.db/kudu_txt",
"defaultFS": "hdfs://myns",
"fileName": "pt=1",
"fieldDelimiter": ",",
"encoding": "utf-8",
"writeMode": "overwrite",
"column": [
{
"name": "id",
"type": "BIGINT"
},
{
"name": "VARCHAR",
"type": "VARCHAR"
},
{
"name": "age",
"type": "BIGINT"
}
],
"hadoopConfig": {
"hadoop.user.name": "root",
"dfs.ha.namenodes.ns": "nn1,nn2",
"fs.defaultFS": "hdfs://myns",
"dfs.namenode.rpc-address.ns.nn2": "hadoop1:9000",
"dfs.client.failover.proxy.provider.ns": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider",
"dfs.namenode.rpc-address.ns.nn1": "hadoop2:9000",
"dfs.nameservices": "myns",
"fs.hdfs.impl.disable.cache": "true",
"fs.hdfs.impl": "org.apache.hadoop.hdfs.DistributedFileSystem"
}
}
}
}
],
"setting" : {
"restore" : {
"restoreColumnName" : "id",
"restoreColumnIndex" : 0
},
"speed" : {
"bytes" : 0,
"readerChannel" : 3,
"writerChannel" : 3
}
}
}
}
启动同步任务 。
bash ./bin/chunjun-standalone.sh -job chunjun-examples/json/mysql/mysql_hdfs_polling_my.json
任务执行完后通过web控制台可以看到执行成功信息,查看HDFS路径数据也可以看到刚刚成功写入的数据 。
创建一个个Kafka的topic用于数据源读取 。
kafka-topics.sh --create --zookeeper zk1:2181,zk2:2181,zk3:2181 --replication-factor 3 --partitions 3 --topic my_test1
ClickHouse创建testdb数据库和sql_side_table表 。
CREATE DATABASE IF NOT EXISTS testdb;
CREATE TABLE if not exists sql_side_table
(
id Int64,
test1 Int64,
test2 Int64
) ENGINE = MergeTree()
PRIMARY KEY (id);
insert into sql_side_table values(1,11,101),(2,12,102),(3,13,103);
MySQL创建sql_sink_table表 。
CREATE TABLE `sql_sink_table` (
`id` bigint NOT NULL,
`name` varchar(100) COLLATE utf8mb4_general_ci DEFAULT NULL,
`test1` bigint DEFAULT NULL,
`test2` bigint DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci
创建sql文件chunjun-examples/sql/clickhouse/kafka_clickhouse_my.sql 。
CREATE TABLE source (
id BIGINT,
name STRING
) WITH (
'connector' = 'kafka-x',
'topic' = 'my_test1',
'properties.bootstrap.servers' = 'kafka1:9092',
'properties.group.id' = 'dodge',
'format' = 'json'
);
CREATE TABLE side (
id BIGINT,
test1 BIGINT,
test2 BIGINT
) WITH (
'connector' = 'clickhouse-x',
'url' = 'jdbc:clickhouse://ck1:8123/testdb',
'table-name' = 'sql_side_table',
'username' = 'default',
'lookup.cache-type' = 'lru'
);
CREATE TABLE sink (
id BIGINT,
name VARCHAR,
test1 BIGINT,
test2 BIGINT
)WITH (
'connector' = 'mysql-x',
'url' = 'jdbc:mysql://mysqlserver:3306/test',
'table-name' = 'sql_sink_table',
'username' = 'root',
'password' = '123456',
'sink.buffer-flush.max-rows' = '1024',
'sink.buffer-flush.interval' = '10000',
'sink.all-replace' = 'true'
);
INSERT INTO sink
SELECT
s1.id AS id,
s1.name AS name,
s2.test1 AS test1,
s2.test2 AS test2
FROM source s1
JOIN side s2
ON s1.id = s2.id
启动同步任务 。
bash ./bin/chunjun-standalone.sh -job chunjun-examples/sql/clickhouse/kafka_clickhouse_my.sql
往kafka的my_test1这个topic写入数据 。
./kafka-console-producer.sh --broker-list cdh1:9092 --topic my_test1
{"id":1,"name":"sunhaiyang"}
{"id":2,"name":"gulili"}
查看MySQL的sql_sink_table表已经有刚才写入消息并关联出结果的数据 。
创建两个Kafka的topic,一个用于数据源读取,一个用于数据源写入 。
kafka-topics.sh --create --zookeeper zk1:2181,zk2:2181,zk3:2181 --replication-factor 3 --partitions 3 --topic my_test3
kafka-topics.sh --create --zookeeper zk1:2181,zk2:2181,zk3:2181 --replication-factor 3 --partitions 3 --topic my_test4
创建sql文件chunjun-examples/sql/kafka/kafka_kafka_my.sql 。
CREATE TABLE source_test (
id INT
, name STRING
, money decimal
, datethree timestamp
, `partition` BIGINT METADATA VIRTUAL -- from Kafka connector
, `topic` STRING METADATA VIRTUAL -- from Kafka connector
, `leader-epoch` int METADATA VIRTUAL -- from Kafka connector
, `offset` BIGINT METADATA VIRTUAL -- from Kafka connector
, ts TIMESTAMP(3) METADATA FROM 'timestamp' -- from Kafka connector
, `timestamp-type` STRING METADATA VIRTUAL -- from Kafka connector
, partition_id BIGINT METADATA FROM 'partition' VIRTUAL -- from Kafka connector
, WATERMARK FOR datethree AS datethree - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka-x'
,'topic' = 'my_test3'
,'properties.bootstrap.servers' = 'kafka1:9092'
,'properties.group.id' = 'test1'
,'scan.startup.mode' = 'earliest-offset'
,'format' = 'json'
,'json.timestamp-format.standard' = 'SQL'
,'scan.parallelism' = '2'
);
CREATE TABLE sink_test
(
id INT
, name STRING
, money decimal
, datethree timestamp
, `partition` BIGINT
, `topic` STRING
, `leader-epoch` int
, `offset` BIGINT
, ts TIMESTAMP(3)
, `timestamp-type` STRING
, partition_id BIGINT
) WITH (
'connector' = 'kafka-x'
,'topic' = 'my_test4'
,'properties.bootstrap.servers' = 'kafka1:9092'
,'format' = 'json'
,'sink.parallelism' = '2'
,'json.timestamp-format.standard' = 'SQL'
);
INSERT INTO sink_test
SELECT *
from source_test;
往kafka的my_test3这个topic写入数据 。
kafka-console-producer.sh --broker-list cdh1:9092 --topic my_test3
{"id":100,"name":"guocai","money":243.18,"datethree":"2023-07-03 22:00:00.000"}
{"id":101,"name":"hanmeimei","money":137.32,"datethree":"2023-07-03 22:00:01.000"}
启动同步任务 。
bash ./bin/chunjun-standalone.sh -job chunjun-examples/sql/kafka/kafka_kafka_my.sql
查看kafka的my_test4的数据,已经收到相应数据并打上kafka元数据信息 。
kafka-console-consumer.sh --bootstrap-server kafka:9092 --topic my_test4 --from-beginning
最后此篇关于国产开源流批统一的数据同步工具Chunjun入门实战的文章就讲到这里了,如果你想了解更多关于国产开源流批统一的数据同步工具Chunjun入门实战的内容请搜索CFSDN的文章或继续浏览相关文章,希望大家以后支持我的博客! 。
实战-行业攻防应急响应 简介: 服务器场景操作系统 Ubuntu 服务器账号密码:root/security123 分析流量包在/home/security/security.pcap 相
背景 最近公司将我们之前使用的链路工具切换为了 OpenTelemetry. 我们的技术栈是: OTLP C
一 同一类的方法都用 synchronized 修饰 1 代码 package concurrent; import java.util.concurrent.TimeUnit; public c
一 简单例子 1 代码 package concurrent.threadlocal; /** * ThreadLocal测试 * * @author cakin */ public class T
1. 问题背景 问题发生在快递分拣的流程中,我尽可能将业务背景简化,让大家只关注并发问题本身。 分拣业务针对每个快递包裹都会生成一个任务,我们称它为 task。task 中有两个字段需要
实战环境 elastic search 8.5.0 + kibna 8.5.0 + springboot 3.0.2 + spring data elasticsearch 5.0.2 +
Win10下yolov8 tensorrt模型加速部署【实战】 TensorRT-Alpha 基于tensorrt+cuda c++实现模型end2end的gpu加速,支持win10、
yolov8 tensorrt模型加速部署【实战】 TensorRT-Alpha 基于tensorrt+cuda c++实现模型end2end的gpu加速,支持win10、linux,
目录如下: 为什么需要自定义授权类型? 前面介绍OAuth2.0的基础知识点时介绍过支持的4种授权类型,分别如下: 授权码模式 简化模式 客户端模式 密码模式
今天这篇文章介绍一下如何在修改密码、修改权限、注销等场景下使JWT失效。 文章的目录如下: 解决方案 JWT最大的一个优势在于它是无状态的,自身包含了认证鉴权所需要的所有信息,服务器端
前言 大家好,我是捡田螺的小男孩。(求个星标置顶) 我们日常做分页需求时,一般会用limit实现,但是当偏移量特别大的时候,查询效率就变得低下。本文将分四个方案,讨论如何优化MySQL百万数
前言 大家好,我是捡田螺的小男孩。 平时我们写代码呢,多数情况都是流水线式写代码,基本就可以实现业务逻辑了。如何在写代码中找到乐趣呢,我觉得,最好的方式就是:使用设计模式优化自己
我们先讲一些arm汇编的基础知识。(我们以armv7为例,最新iphone5s上的64位暂不讨论) 基础知识部分: 首先你介绍一下寄存器: r0-r3:用于函数参数及返回值的传递 r4-r6
一 同一类的静态方法都用 synchronized 修饰 1 代码 package concurrent; import java.util.concurrent.TimeUnit; public
DRF快速写五个接口,比你用手也快··· 实战-DRF快速写接口 开发环境 Python3.6 Pycharm专业版2021.2.3 Sqlite3 Django 2.2 djangorestfram
一 添加依赖 org.apache.thrift libthrift 0.11.0 二 编写 IDL 通过 IDL(.thrift 文件)定义数据结构、异常和接口等数据,供各种编程语言使用 nam
我正在阅读 Redis in action e-book关于semaphores的章节.这是使用redis实现信号量的python代码 def acquire_semaphore(conn, semn
自定义控件在WPF开发中是很常见的,有时候某些控件需要契合业务或者美化统一样式,这时候就需要对控件做出一些改造。 目录 按钮设置圆角
师父布置的任务,让我写一个服务练练手,搞清楚socket的原理和过程后跑了一个小demo,很有成就感,代码内容也比较清晰易懂,很有教育启发意义。 代码 ?
? 1 2
我是一名优秀的程序员,十分优秀!