开源数据交换(client)

发布于: 2 小时前
开源数据交换(client)

一.项目简介

exchange是要打造一个轻量级,高扩展性的数据交换平台,支持对结构化及无结构化的异构数据源之间的数据传输,在应用层上具有数据权限管控、节点服务高可用和多租户资源隔离等业务特性,而在数据层上又具有传输架构多样化、模块插件化和组件低耦合等架构特点。

exchange的传输能力依赖于Apache Beam链路计算的能力,再由事件模型扩展并发能力,最后处理成DAG应用,可以分发到不同的引擎上。

URL

https://github.com/77954309/exchange-parent

客户端

类似与Sqoop操作,核心功能导入/导出,把每个操作基于事件模型并发处理,根据Beam计算模型生成DAG应用。

服务端

可视化操作,传输交换能力依赖于其底层聚合的传输引擎,其顶层对各类数据源定义统一的参数模型,每种传输引擎对参数模型进行映射配置,转化为引擎的输入模型,可视化操作。(尚未开源)

核心特点

  • I/O 高效扩展性(根据Beam规范,扩展)。

  • 数据源管理,目前支持Mysql,Oracle,Hive,Neo4j,Elasticsearch,Gbase,File。

  • 多传输引擎支持(扩展引擎),目前支持Java,Spark,Flink引擎。

  • 近实时任务管控

  • 支持无结构化传输

  • 任务状态自检

  • 各个源根据事件互通传输

教程

[Beam官网](https://beam.apache.org/)

[Apache Beam 大数据处理一站式分析](https://gitbook.cn/gitchat/activity/5dad728e7c3fea79dbc619a4)

二.编译部署

2.1 客户端

环境准备

JDK (1.8.0_141) 必选

Spark (2.x) 可选

Flink (1.6.x) 可选

Hadoop 2.7以上

Hive 1.2.1以上

若需要 oralce 把pom注释打开

若需要 gbase 把pom注释打开

如不满足,可以修改pom文件,重新编译

编译

当前项目环境下编译

示例:D:\\exchange-parent> mvn clean package

执行文件

kd-process-0.1.0-shaded.jar

三.快速使用

3.1 客户端

具体参数含义,请看开发规范

spark2-submit  --driver-memory 2G --executor-memory 5G --executor-cores 3 --num-executors 3  --conf spark.default.para
llelism=200 --conf spark.sql.shuffle.partitions=200  --class com.knowlegene.parent.process.SwapSparkApplication  --mas
ter yarn   /mnt/jars/kd-process-0.1.0-shaded.jar  --fromName=mysql --toName=hive --hiveClass=org.apache.hive.jdbc.Hive
Driver --hiveUrl=jdbc:hive2://192.168.200.117:10000/linkis_db  --hiveUsername=hdfs --hivePassword=hdfs --hiveTableName
=per_test1 --hiveTableEmpty=true  --url=jdbc:mysql://192.168.100.101:3306/dt?useSSL=false --tableName=companykeyrate -
-driverClass=com.mysql.jdbc.Driver --username=dt --password=Dt@654321  --runner=SparkRunner  --sparkMaster=yarn  

四.底层已支持计算存储引擎

4.1 客户端

Spark

com.knowlegene.parent.process.SwapSparkApplication 这个时入口程序,

--runner=SparkRunner  --sparkMaster=yarn  这两个Beam参数必传,具体用Spark的yarn,local,standalone 看需求。具体操作规范请看Beam(https://beam.apache.org/documentation/)。

spark2-submit --driver-memory 2G --executor-memory 5G --executor-cores 3 --num-executors 3 --conf spark.default.para
llelism=200 --conf spark.sql.shuffle.partitions=200 --class com.knowlegene.parent.process.SwapSparkApplication --mas
ter yarn /mnt/jars/kd-process-0.1.0-shaded.jar --fromName=mysql --toName=hive --hiveClass=org.apache.hive.jdbc.Hive
Driver --hiveUrl=jdbc:hive2://192.168.200.117:10000/linkis_db --hiveUsername=hdfs --hivePassword=hdfs --hiveTableName
=per_test1 --hiveTableEmpty=true --url=jdbc:mysql://192.168.100.101:3306/dt?useSSL=false --tableName=companykeyrate -
-driverClass=com.mysql.jdbc.Driver --username=dt --password=Dt@654321 --runner=SparkRunner --sparkMaster=yarn

Java

com.knowlegene.parent.process.SwapDirectApplication 这个时入口程序,

根据java -cp 传参运行,具体操作规范请看Beam(https://beam.apache.org/documentation/)

Flink

com.knowlegene.parent.process.SwapFlinkApplication 这个时入口程序,

mvn exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
-Pflink-runner \
-Dexec.args="--runner=FlinkRunner \
--inputFile=/path/to/pom.xml \
--output=/path/to/counts \
--flinkMaster=<flink master url> \
--filesToStage=target/word-count-beam-bundled-0.1.jar"

具体操作规范请看Beam(https://beam.apache.org/documentation/)

五.架构

客户端

支持数据源

六.开发规范

6.1 客户端传参规范

入口

| 参数 | 含义 |

|:----|:----|

| fromName | 起始数据源(hive,oracle,mysql,es,file,gbase,neo4j) |

| toName | 目标数据源(hive,oracle,mysql,es,file,gbase,neo4j) |

| runner | 引擎名称(SparkRunner,FlinkRunner) |

更多引擎参数

FlinkPipelineOptions

SparkPipelineOptions

Hive

| 参数 | 含义 |

|:----|:----|

| hiveClass | org.apache.hive.jdbc.HiveDriver |

| hiveUrl | hive jdbc url |

| hiveUsername | 用户名 |

| hivePassword | 密码 |

| hiveTableName | 表名称 |

| hiveDatabase | 库名称 |

| hiveTableEmpty | 是否清空目标表数据,默认不清空(false) |

| hiveSQL | hive sql |

| hiveColumn | 起始,目标,元数据的列字段 |

| hMetastoreHost | Hcatalog host |

| hMetastorePort | Hcatalog port |

| hiveFilter | Hcatalog 过滤条件 |

| hivePartition | 分区,json ,示例({“time”:"2019"}) |

| hiveUrls | 起始源和目标源都是Hive,url数组 |

| hiveUsernames | 起始源和目标源都是Hive,用户数组 |

| hivePasswords | 起始源和目标源都是Hive,密码数组 |

| hiveTableNames | 起始源和目标源都是Hive,表名称数组 |

| hiveDatabases | 起始源和目标源都是Hive,库名称数组 |

| hMetastoreHosts | 起始源和目标源都是Hive,Hcatalog host数组 |

| hMetastorePorts | 起始源和目标源都是Hive,Hcatalog port数组 |

Mysql

| 参数 | 含义 |

|:----|:----|

| url | jdbc url |

| tableName | 表名称 |

| username | 用户名称 |

| password | 密码 |

| driverClass | driver class |

| dbSQL | sql |

| dbColumn | 起始,目标,元数据的列字段 |

| driverClasss | 起始源和目标源都是db |

| urls | 起始源和目标源都是db |

| tableNames | 起始源和目标源都是db |

| usernames | 起始源和目标源都是db |

| passwords | 起始源和目标源都是db |

Elasticsearch

| 参数 | 含义 |

|:----|:----|

| esAddrs | es 地址 |

| esIndex | 索引 |

| esType | 类型,7版本必须_doc |

| esQuery | 查询 |

| esIdFn | 根据字段赋值_id |

| esAddrsFrom | 起始源和目标源都是es |

| esAddrsTo | 起始源和目标源都是es |

| esIndexs | 起始源和目标源都是es |

| esTypes | 起始源和目标源都是es |

| nestingKeysName | 嵌套名称 |

| nestingKeys | 根据key嵌套 |

| nestingValues | 嵌套数组中value |

Neo4j

| 参数 | 含义 |

| :---- | :---- |

| cypher | cypher |

| neoUrl | url |

| neoUsername | 用户名称 |

| neoPassword | 密码 |

| neoFormat | 导出模板,具体看测试用例 |

| neoType | 类型,起始数据源为Neo4j,节点(node)/关系(relate) |

| cyphers | 起始源和目标源都是neo4j |

| neoUrls | 起始源和目标源都是neo4j |

| neoUsernames | 起始源和目标源都是neo4j |

| neoPasswords | 起始源和目标源都是neo4j |

Oracle

| 参数 | 含义 |

|:----|:----|

| url | jdbc url |

| tableName | 表名称 |

| username | 用户名称 |

| password | 密码 |

| driverClass | driver class |

| dbSQL | sql |

| dbColumn | 起始,目标,元数据的列字段 |

| driverClasss | 起始源和目标源都是db |

| urls | 起始源和目标源都是db |

| tableNames | 起始源和目标源都是db |

| usernames | 起始源和目标源都是db |

| passwords | 起始源和目标源都是db |

Gbase

| 参数 | 含义 |

|:----|:----|

| url | jdbc url |

| tableName | 表名称 |

| username | 用户名称 |

| password | 密码 |

| driverClass | driver class |

| dbSQL | sql |

| dbColumn | 起始,目标,元数据的列字段 |

| driverClasss | 起始源和目标源都是db |

| urls | 起始源和目标源都是db |

| tableNames | 起始源和目标源都是db |

| usernames | 起始源和目标源都是db |

| passwords | 起始源和目标源都是db |

File

| 参数 | 含义 |

|:----|:----|

| filePath | 文件路径 |

| fieldDelim | 文件分隔符 |

| fieldTitle | 文件字段名称 |

| filePaths | 起始源和目标源都是file |

| fieldDelims | 起始源和目标源都是file |

6.2 测试用例

参考

com.knowlegene.parent.process 里面包括所有源测试用例

七.问题

1.Hive版本1.2以上,不然只能使用jdbc,无法使用HCatalog

2.Neo4j 多Type用模板方式导出,否则建议用cypher,具体情况,看测试用例

3.Gbase,Oracle jar 自行引入,pom中注释了

4.Neo4j 模板方式导出,id:ID(Node) 固定,node表示节点名称。

5.es的嵌套当前版本只支持一个。

6.有的引擎若用不到自行删除。

7.推荐HCatalog。

8.Hive jdbc 性能不好,不建议用java引擎,用Spark/Flink。

9.关于Beam教程https://gitbook.cn/gitchat/activity/5dad728e7c3fea79dbc619a4

八.总结

后续继续开源,欢迎交流,欢迎Star/Fork。

https://github.com/77954309/exchange-parent

发布于: 2 小时前 阅读数: 14
用户头像

李孟

关注

还未添加个人签名 2017.10.18 加入

数据工程师 https://limeng.blog.csdn.net/

评论

发布
暂无评论
开源数据交换(client)