精选文章 flinkx学习

flinkx学习

作者:wandy0211 时间: 2019-11-06 11:29:03
wandy0211 2019-11-06 11:29:03

什么是FlinkX

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

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

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

3 快速起步

3.1 运行模式

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

3.2 执行环境

  • Java: JDK8及以上
  • Flink集群: 1.4及以上(单机模式不需要安装Flink集群)
  • 操作系统:理论上不限,但是目前只编写了shell启动脚本,用户可以可以参考shell脚本编写适合特定操作系统的启动脚本。

3.3 打包

进入项目根目录,使用maven打包:

mvn clean package -Dmaven.test.skip

打包结束后,项目根目录下会产生bin目录和plugins目录,其中bin目录包含FlinkX的启动脚本,plugins目录下存放编译好的数据同步插件包

3.4 启动

3.4.1 命令行参数选项

 model

  • 描述:执行模式,也就是flink集群的工作模式
    • local: 本地模式
    • standalone: 独立部署模式的flink集群
    • yarn: yarn模式的flink集群,需要提前在yarn上启动一个flink session,使用默认名称"Flink session cluster"
  • 必选:否
  • 默认值:local

job

  • 描述:数据同步任务描述文件的存放路径;该描述文件中使用json字符串存放任务信息。
  • 必选:是
  • 默认值:无

pluginRoot

  • 描述:插件根目录地址,也就是打包后产生的pluginRoot目录。
  • 必选:是
  • 默认值:无

flinkconf

  • 描述:flink配置文件所在的目录(单机模式下不需要),如/hadoop/flink-1.4.0/conf
  • 必选:否
  • 默认值:无

yarnconf

  • 描述:Hadoop配置文件(包括hdfs和yarn)所在的目录(单机模式下不需要),如/hadoop/etc/hadoop
  • 必选:否
  • 默认值:无

3.4.2 启动数据同步任务

  • 以本地模式启动数据同步任务
bin/flinkx -mode local -job /Users/softfly/company/flink-data-transfer/jobs/task_to_run.json -pluginRoot /Users/softfly/company/
  • 以standalone模式启动数据同步任务
bin/flinkx -mode standalone -job /Users/softfly/company/flink-data-transfer/jobs/oracle_to_oracle.json  -pluginRoot /Users/softfly/company/flink-data-transfer/plugins -flinkconf /hadoop/flink-1.4.0/conf -confProp "{"flink.checkpoint.interval":60000,"flink.checkpoint.stateBackend":"/flink_checkpoint/"}" -s /flink_checkpoint/0481473685a8e7d22e7bd079d6e5c08c/chk-*

 

4.1 setting

    "setting": {
        "speed": {...},
        "errorLimit": {...},
        "dirty": {...}
    }

setting包括speed、errorLimit和dirty三部分,分别描述限速、错误控制和脏数据管理的配置信息

4.1.1 speed

            "speed": {
                 "channel": 3,
                 "bytes": 0
            }

 

4.1.2 errorLimit

            "errorLimit": {
                "record": 10000,
                "percentage": 100
            }

 

4.1.3 dirty

        "dirty": {
                "path": "/tmp",
                "hadoopConfig": {
                    "fs.default.name": "hdfs://ns1",
                    "dfs.nameservices": "ns1",
                    "dfs.ha.namenodes.ns1": "nn1,nn2",
                    "dfs.namenode.rpc-address.ns1.nn1": "node02:9000",
                    "dfs.namenode.rpc-address.ns1.nn2": "node03:9000",
                    "dfs.ha.automatic-failover.enabled": "true",
                    "dfs.client.failover.proxy.provider.ns1": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider",
                    "fs.hdfs.impl.disable.cache": "true"
                }
            }

 

4.1.4 restore

"restore": {

        "isRestore": false,
        "restoreColumnName": "",
        "restoreColumnIndex": 0
      }

restore配置请参考断点续传

4.2 content

        "content": [
            {
               "reader": {
                    "name": "...",
                    "parameter": {
                        ...
                    }
                },
               "writer": {
                    "name": "...",
                    "parameter": {
                         ...
                     }
                }
            }
        ]

 

reader和writer包括name和parameter,分别表示插件名称和插件参数

  • 以yarn模式启动数据同步任务
  • bin/flinkx -mode yarn -job /Users/softfly/company/flinkx/jobs/mysql_to_mysql.json  -pluginRoot /opt/dtstack/flinkplugin/syncplugin -flinkconf /opt/dtstack/myconf/conf -yarnconf /opt/dtstack/myconf/hadoop -confProp "{"flink.checkpoint.interval":60000,"flink.checkpoint.stateBackend":"/flink_checkpoint/"}" -s /flink_checkpoint/0481473685a8e7d22e7bd079d6e5c08c/chk-*
    

    4 数据同步任务模版

  • 从最高空俯视,一个数据同步的构成很简单,如下:

    {
        "job": {
            "setting": {...},
            "content": [...]
        }
    }
    

    数据同步任务包括一个job元素,而这个元素包括setting和content两部分。

  • setting: 用于配置限速、错误控制和脏数据管理
  • content: 用于配置具体任务信息,包括从哪里来(Reader插件信息),到哪里去(Writer插件信息)
  • channel: 任务并发数
  • bytes: 每秒字节数,默认为 Long.MAX_VALUE
  • record: 出错记录数超过record设置的条数时,任务标记为失败
  • percentage: 当出错记录数超过percentage百分数时,任务标记为失败
  • path: 脏数据存放路径
  • hadoopConfig: 脏数据存放路径对应hdfs的配置信息(hdfs高可用配置)
  • reader: 用于读取数据的插件的信息
  • writer: 用于写入数据的插件的信息

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

 

勿删,copyright占位
分享文章到微博
分享文章到朋友圈

上一篇:Mac OS下安装pygame一直出错,终于解决!

下一篇:【二叉树(二)】:二叉搜索树

您可能感兴趣

  • Java学习笔记2

    Java 第二天 一、类型转换 1. 隐式转换(从小到大) byte,short,char –> int –> long –> float –>double 2. 强制转换(从大到小) 格式:目标数据类型 变量名 = (目标数据类型)(被转换的数据) 3. 相关问题 public class transform { public static void main(String[] ...

  • pytorch triple-loss

    一、Triplet结构: triplet loss是一种比较好理解的loss,triplet是指的是三元组:Anchor、Positive、Negative: 整个训练过程是: 首先从训练集中随机选一个样本,称为Anchor(记为x_a)。 然后再随机选取一个和Anchor属于同一类的样本,称为Positive (记为x_p) 最后再随机选取一个和Anchor属于不同类的样本,称为Negat...

  • SIP语音环境中十大经典问题及解决办法

    在VOIP的环境中,特别是基于SIP通信的环境中,我们经常会遇到一些非常常见的问题,例如,单通,30秒就断线,注册问题,回声等。这些问题事实上都有非常直接的排查方式和解决办法,用户可以按照一定的排查方式,工具非常高效地解决这些问题。但是,因为读者技术水平参差不齐,网络上的很多技术也不完整。笔者今天系统归纳了这些问题。根据一些用户的使用环境和用户经常遇到一些问题,我们列举了以下十个在SIP呼叫...

  • SpringMVC框架的入门学习01

    SpringMVC的基本概念 关于三层架构和MVC 三层架构 我们的开发架构一般都是基于两种形式,一种是 C/S 架构,也就是客户端/服务器,另一种是 B/S 架构,也就是浏览器服务器。在 JavaEE 开发中,几乎全都是基于 B/S架构的开发。那么在 B/S架构中,系统标准的三层架构包括:表现层、业务层、持久层。三层架构在我们的实际开发中使用的非常多,所以很多案例也都是基于三层架构设计的。...

  • 学习HTML5大前端可以从事哪些工作?

      近几年,随着互联网技术的日渐成熟,越来越多的年轻人想往互联网行业发展,就单纯的HTML5大前端网页设计师这一岗位就吸引了无数人。虽然HTML5大前端网页设计门槛相对较低,但是薪资却很高。当然,除了HTML5大前端网页设计外,还有其他的岗位可供选择。今天就让千锋为大家分享一下学习HTML5大前端可以从事的工作有哪些?希望对大家选择相应岗位有所帮助。   一、HTML5开发工程师   HTM...

  • 零基础学习HTML5前端提升攻略!

      学习一门技术,为未来蓄积能量。那么我们可以学习的技术有哪些呢?当然每个人的兴趣点不一样,选择的方向也就不同,今天,我们就为零基础的小白提供学习HTML5大前端提升攻略。   首先,应该重视基础的重要性。无论做什么事情都一定要有扎实的基础,学习HTML5大前端也不例外。只有基础牢固,才能更深入的学习新技术、新知识。HTML5大前端的入门门槛很低,与其他语言先慢后快的学习节奏相比,它是一个先...

  • 2020B证(安全员)考试及B证(安全员)模拟考试题库

    题库来源:安全生产模拟考试一点通公众号小程序 2020B证(安全员)考试及B证(安全员)模拟考试题库,包含B证(安全员)考试答案解析及B证(安全员)模拟考试题库练习。由安全生产模拟考试一点通公众号结合国家B证(安全员)考试最新大纲及B证(安全员)考试真题出具,有助于B证(安全员)考试软件考前练习。 1、【单选题】多台挖掘机在同一作用面机械开挖,挖掘机间距应大于( )。( B ) A、5m B...

  • OpenCV 识别图片中的米粒个数,并计算米粒的平均面积和长度(转)

    介绍 OpenCV+Python 使用OpenCV构建图像识别算法,识别图片中的米粒个数,并计算米粒的平均面积和长度 软件架构 模块:OpenCV 4.0.0.21 编程语言:Python 3.7.2 编译器:PyCharm 2018 程序设计思路 首先介绍一下程序设计的思路: 图像采集(取到图像):可以用摄像头拍摄或者图片直接导入 图像预处理:对图像进行灰度化 基于灰度的阈值分割:使用局部...

华为云40多款云服务产品0元试用活动

免费套餐,马上领取!
CSDN

CSDN

中国开发者社区CSDN (Chinese Software Developer Network) 创立于1999年,致力为中国开发者提供知识传播、在线学习、职业发展等全生命周期服务。