devbox
空白诗007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

大数据 Flink SQL CDC 实时数据整库同步 实时宽表 实时计算 流批一体_flink cdc 必须使用hdfs吗

作者：空白诗007 | 2024-08-05 21:42:46

赞

踩

flink cdc 必须使用hdfs吗

目录

应用场景

在这里插入图片描述

Apache Flink 是 Apache 软件基金会的顶级项目，也是当下被广泛使用的开源大数据计算引擎之一。基于它 “流批一体” 的技术，越来越多的企业选择 Apache Flink 应用于自身的业务场景，如数据集成、数据分析、数据仓库、实时分析、实时大屏等场景中，解决实时计算的需求。近年来，Apache Flink 开始广泛应用于推荐、广告和搜索等机器学习业务场景，已覆盖近百家企业的绝大多数实时计算需求，包括互联网娱乐、游戏、电商、金融、证劵、通信等多个行业。
目前 Flink 基本服务于阿里的所有 BU ，在双十一峰值的计算能力达到 40 亿条每秒，计算任务达到了 3 万多个，总共使用 100 万+ Core ；几乎涵盖了集团内的所有具体业务，比如：数据中台、AI 中台、风控中台、实时运维、搜索推荐等。

优势

流批一体
高吞吐、低延迟
exactly-once 的状态一致性保证
结果的准确性和良好的容错性
可以众多常用的存储系统连接
高可用，支持动态扩展

应用案例

基于 Apache Flink 和规则引擎的实时风控解决方案 ​

日均百亿级日志处理：微博基于Flink的实时计算平台建设

基于Flink的超大规模在线实时反欺诈系统的建设与实践

每秒1.8亿条高峰流量，美团数仓上Flink

Apache Flink X Apache Doris 构建极速易用的实时数仓架构

中原银行 OLAP 架构实时化演进

芒果 TV 基于 Flink 的实时数仓建设实践

Flink 使用教程

Flink 从 0 到 1 学习 —— Flink Data transformation(转换)

第十四篇|Flink SQL之维表join

Flink run 参数解释

Flink与Nacos集成

Flink多sink输出

Flink 开源项目系统

flink-recommandSystem-demo: 基于Flink实现的商品实时推荐系统

flink-userportrait-main: 基于Flink流处理的动态实时亿级全端用户画像系统

Flink 电商用户行为统计 流量统计 近期热门商品统计 用户画像分析 商品推荐 风控

flink-streaming-platform-web: 基于flink-sql的实时流计算web平台

Flink SQL

文档

Flink SQL 官网文档

flink sql大量使用案例

如何实时统计最近 15 秒的商品销售额

Flink 系例之 SQL 案例 - 游戏上报数据统计

flink sql 计算实时计算uv和pv

使用Flink SQL将数据写入到ClickHouse

基于 FLINK SQL 的实时数据打宽

flink sql 示例

flink sql kafka canal-json source

Flink 如何将聚合结果写入kafka ——读写canal-json

vi products_kafka_to_doris.sql
SET parallelism.default = 1;
SET execution.checkpointing.interval = 3s;
SET sql-client.execution.result-mode = tableau;

CREATE TABLE products_kafka_source (
  id INT,
  name STRING,
  description STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'fms_test_mysql_topic',
  'properties.bootstrap.servers' = '10.18.2.221:9094',
  'properties.group.id' = 'flink_kafka_test',
  'format' = 'canal-json',
  'scan.startup.mode' = 'latest-offset'
);

CREATE TABLE products_doris_sink (
  id INT,
  name STRING,
  description STRING
) WITH (
'connector' = 'doris',
'fenodes' = '10.18.2.23:8030',
'benodes' = '10.18.2.23:8040',
'table.identifier' = 'app_db.ods_products',
'username' = 'root',
'password' = ''
);

INSERT INTO products_doris_sink SELECT id, name, description FROM products_kafka_source;

启动 ./bin/sql-client.sh -f products_kafka_to_doris.sql


source 打印
drop table if exists print_table;
CREATE TABLE print_table WITH ('connector' = 'print')
LIKE products_kafka_source (EXCLUDING ALL)

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40

时间窗口

窗口函数 | Apache Flink 官网

FlinkSql的窗口使用以及运用案例

FlinkSql中的窗口

FlinkSql开窗实例：消费kafka写入文本

10分钟了解Flink窗口计算 非sql

Flink SQL流表与维表join 和 双流join

Flink SQL 实战：双流 join 场景应用

Flink SQL 双流 join demo

Flink SQL流表与维表join 和 双流join

Flink-join(流表关联流表，流表关联维表),自定义函数

Flink SQL实践记录4 – 实时更新的维表如何join

Flink CDC

Apache Flink CDC 官网

使用 Flink CDC 构建 Streaming ETL | Apache Flink CDC

基于 Flink CDC 实现海量数据的实时同步和转换

Flink CDC，实现Mysql数据增量备份到Clickhouse

Flink CDC采集MySQL binlog日志实时写入ClickHouse

flink cdc mysql --> elasticsearch

Flink SQL CDC 上线！我们总结了 13 条生产实践经验

flink cdc mysql --> apache doris

flink-cdc官网 MySQL 同步到 Doris | Apache Flink CDC
测试脚本

注意事项：
flink 启动配置修改 conf/flink-conf.yaml
jobmanager.bind-host: 改成 0.0.0.0
taskmanager.bind-host: 改成 0.0.0.0
taskmanager.numberOfTaskSlots: 改成 5
parallelism.default: 改成 10
增加 execution.checkpointing.interval: 3000
rest.bind-address: 改成 0.0.0.0
web.cancel.enable: 改成 true

mysql库名和doris库名不相同时必须得配置route参数 必须mysql表与doris表一对一配置。如果mysql有的表但route没配置cdc也会报错

vi mysql-to-doris.yaml
source:
  type: mysql
  hostname: localhost
  port: 5985
  username: root
  password: test123244
  tables: flink_test.\.*
  server-id: 5400-5404
  server-time-zone: UTC

sink:
  type: doris
  fenodes: 10.18.25.21:8030
  benodes: 10.18.25.21:8040
  username: root
  password: ""
  table.create.properties.light_schema_change: true
  table.create.properties.replication_num: 1

route:
  - source-table: flink_test.orders
    sink-table: app_db.ods_orders
  - source-table: flink_test.shipments
    sink-table: app_db.ods_shipments
  - source-table: flink_test.products
    sink-table: app_db.ods_products

pipeline:
  name: Sync MySQL Database to Doris
  parallelism: 2

启动脚本
bash bin/flink-cdc.sh mysql-to-doris.yaml --jar lib/mysql-connector-java-8.0.27.jar --flink-home /data0/flink/flink-1.18.0

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46

flink cdc 实时统计计算

统计mysql 订单表每天的gmv并sink到kafka

mysql --> Canal --> kafka --> Flink --> HDFS

基于Canal与Flink实现数据实时增量同步(一)

基于Canal与Flink实现数据实时增量同步(二)

Dinky

Dinky官网: Dinky 是一个开箱即用的一站式实时计算平台，以 Apache Flink 为基础，连接 OLAP 和数据湖等众多框架，致力于流批一体和湖仓一体的建设与实践。

docker 部署

docker run -d --restart=always --net=host -e MYSQL_ADDR=m3309i.grid.xxxx.com.cn:3309 -e MYSQL_DATABASE=dlink -e MYSQL_USERNAME=mis -e MYSQL_PASSWORD=12434qwer --name dinky dinkydocker/dinky-standalone-server:0.7.5-flink16

docker cp dinky:/opt/dinky/plugins .

docker run增加 -v /data0/dinky/plugins:/opt/dinky/plugins

  • 1
  • 2
  • 3
  • 4
  • 5

基于 Dinky + FlinkSQL + Flink CDC 同步 MySQL 数据到 Elasticsearch、Kafka

Flink部署

flink flink-cdc 相关jar包

flink flink-cdc jar
flink-sql-connector-kafka jar
flink-sql-connector-mysql-cdc jar
flink-doris-connector jar

不带sql的jar包 需手动引入第三方依赖jar包 比如kafka clients jar 版本要对应
flink-connector-kafka-1.17.2.jar
kafka-clients-3.2.3.jar
kafka-clients jar

flink sql mysql jdbc相关jar包
flink-connector-jdbc-3.0.0-1.16.jar
注意：新增jar包记得重启

docker 部署flink

docker pull flink:1.17.0

# 创建 docker 网络，方便 JobManager 和 TaskManager 内部访问
 docker network create flink-network
 
# 创建 JobManager 
 docker run \
  -itd \
  --name=jobmanager \
  --publish 8081:8081 \
  --network flink-network \
  --env FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager" \
  flink:1.16.0 jobmanager 
  
# 创建 TaskManager 
 docker run \
  -itd \
  --name=taskmanager \
  --network flink-network \
  --env FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager" \
  flink:1.16.0 taskmanager 

mkdir /data0/flink
docker cp jobmanager:/opt/flink/conf conf
docker cp jobmanager:/opt/flink/lib lib

# 启动 jobmanager   
docker run -itd  -v /data0/flink/conf/:/opt/flink/conf/ -v /data0/flink/lib/:/opt/flink/lib/   --name=jobmanager --publish 8081:8081 --env FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager" --network flink-network flink:1.16.0 jobmanager
# 启动 taskmanager   
docker run -itd  -v /data0/flink/conf/:/opt/flink/conf/ -v /data0/flink/lib/:/opt/flink/lib/ --name=taskmanager --network flink-network --env FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager"  flink:1.16.0 taskmanager

docker exec -it jobmanager ./bin/sql-client.sh

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32

docker-compose 部署flink

Docker Hub flink最新版本

docker安装flink

使用Docker搭建Flink集群

部署命令：

cd /data0
mkdir flink
mkdir flink/conf
mkdir flink/lib

#docker pull flink:1.14.4-scala_2.12
docker pull flink:1.13.6-scala_2.11

vi docker-compose.yml
version: "2.1"
services:
  jobmanager:
    image: flink:1.13.6-scala_2.11
    expose:
      - "6123"
    ports:
      - "8081:8081"
    command: jobmanager
    #volumes:
      #- /data0/flink/lib:/opt/flink/lib
      #- /data0/flink/conf:/opt/flink/conf
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager
  taskmanager:
    image: flink:1.13.6-scala_2.11
    expose:
      - "6121"
      - "6122"
    depends_on:
      - jobmanager
    command: taskmanager
    links:
      - "jobmanager:jobmanager"
    environment:
      - JOB_MANAGER_RPC_ADDRESS=jobmanager
    #volumes:
      #- /data0/flink/lib:/opt/flink/lib
      #- /data0/flink/conf:/opt/flink/conf

启动：
docker-compose build
docker-compose up -d --force-recreate
docker-compose down
docker-compose restart

查看日志：
docker logs --tail="600" flink_jobmanager_1
docker logs -f flink_taskmanager_1

配置 flink-sql-connector-mysql-cdc-2.2.0.jar:
docker cp lib/flink-sql-connector-mysql-cdc-2.2.0.jar flink_jobmanager_1:/opt/flink/lib

docker cp flink_jobmanager_1:/opt/flink/conf /data0/flink/
docker cp flink_jobmanager_1:/opt/flink/lib /data0/flink/

打开docker-compose.yml注释内容 配置本地文件映射路径
volumes:
  - /data0/flink/lib:/opt/flink/lib
  - /data0/flink/conf:/opt/flink/conf

docker-compose up -d --force-recreate

使用flink sql-client:
docker exec -it flink_jobmanager_1 /bin/bash
./bin/sql-client.sh
或：
docker exec -it flink_jobmanager_1 ./bin/sql-client.sh

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67

mac 安装flink 指定版本

下载flink 找到需要的版本进行下载
https://www.apache.org/dyn/closer.lua/flink/

我下载的版本：
https://dlcdn.apache.org/flink/flink-1.13.6/flink-1.13.6-bin-scala_2.11.tgz

vi ~/.bash_profile中追加

alias flink="/Users/zhaoyang10/Documents/flink/flink-1.13.6/bin/flink"

  • 1

终端执行：
flink --version

启动flink

cd /Users/zhaoyang10/Documents/flink/flink-1.13.6/bin

  • 1

启动命令

./start-cluster.sh

  • 1

停止命令

./stop-cluster.sh

  • 1

执行job任务：

flink run -d /Users/zhaoyang10/Documents/git/bop-data-sync/data-sync-flink/target/data-sync-flink-1.0.0.jar

  • 1

flink 日志实时监控告警系统

在这里插入图片描述

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/空白诗007/article/detail/934411
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号