seatunnel 简单使用(原名waterdrop)

业务背景

将hive中多个表数据同步到clickhouse中提供实时查询，表均2亿条记录。对同步工具的要求一是能够实现抽数时间不宜过长；二是能够自定义控制将数据抽取到clickhouse集群指定的节点实例上。作为一名java开发，自然不想过多依赖Hadoop那一套，网上搜索一番后决定使用seatunnel，通过简单配置化就可以实现数据的抽取。

简介

Apache SeaTunnel （Incubating) 是一个分布式、高性能、易扩展、用于海量数据（离线&实时）同步和转化的数据集成平台。

官方文档：https://interestinglab.github.io/seatunnel-docs/#/

安装

安装比较简单，参考官方文档即可。

配置

config.conf 下述配置是从hive中抽数插入到clickhouse中的配置，数据源是hive的一张表，通过seatunnel插件根据id字段进行分片插入clickhouse集群不同分片。

spark {spark.sql.catalogImplementation = "hive"spark.app.name = "hive2clickhouse"spark.executor.instances = 30spark.executor.cores = 1 spark.executor.memory = "2g"spark.ui.port = 13000
}input {hive {pre_sql = "select id,name,create_time from table"table_name = "table_tmp"}
}filter {convert {source_field = "data_source"new_type = "UInt8"}org.interestinglab.waterdrop.filter.Slice {source_table_name = "table_tmp"source_field = "id"slice_num = 2slice_code = 0result_table_name = "table_8123"}org.interestinglab.waterdrop.filter.Slice {source_table_name = "table_tmp"source_field = "id"slice_num = 2slice_code = 1result_table_name = "table_8124"}
}output {clickhouse {source_table_name="table_8123"host = "ip1:8123"database = "db_name"username="username"password="pwd"table = "table1"fields = ["id","name","create_time"]clickhouse.socket_timeout = 50000retry_codes = [209, 210]retry = 3bulk_size = 500000}clickhouse {source_table_name="table_8124"host = "ip2:8124"database = "db_name"username="username"password="pwd"table = "table1"fields = ["id","name","create_time"]clickhouse.socket_timeout = 50000retry_codes = [209, 210]retry = 3bulk_size = 500000}
}

插件开发

package org.interestinglab.waterdrop.filterimport io.github.interestinglab.waterdrop.apis.BaseFilter
import io.github.interestinglab.waterdrop.config.{Config, ConfigFactory}
import org.apache.spark.sql.functions.{col, hash, lit, udf}
import org.apache.spark.sql.{Dataset, Row, SparkSession}class Slice extends BaseFilter {var conf: Config = ConfigFactory.empty()/*** Set Config.* */override def setConfig(config: Config): Unit = {this.conf = config}/*** Get Config.* */override def getConfig(): Config = {this.conf}override def checkConfig(): (Boolean, String) = {if (!conf.hasPath("source_field")) {(false, "please specify [source_field] as a non-empty string")} else if (!conf.hasPath("slice_code")) {(false, "please specify [slice_code] as a non-empty string")} else if (!conf.hasPath("slice_num")) {(false, "please specify [slice_num] as a non-empty string")} else {(true, "")}}override def process(spark: SparkSession, df: Dataset[Row]): Dataset[Row] = {val srcField = conf.getString("source_field")val sliceCode = conf.getInt("slice_code")val sliceNum = conf.getInt("slice_num")df.filter(func(hash(col(srcField)), lit(sliceNum), lit(sliceCode)))}val func = udf((s: String, num: Int, target: Int) => {val moCOde = s.toDouble % numval absValue = moCOde.toInt.absabsValue == target})
}

启动

../bin/start-waterdrop.sh --master local[4] --deploy-mode client --config.conf

seatunnel 简单使用(原名waterdrop)相关推荐

Apache SeaTunnel (Incubating) 2.1.0 发布，内核重构、全面支持 Flink
点击蓝字关注我们 2021 年 12 月 9 日,SeaTunnel (原名 Waterdrop) 进入 Apache 孵化器.4 个月后的 2022 年 3 月 18 日,其首个 Apache 版 ...
数据同步工具—SeaTunnel简介
SeaTunnel 简介 SeaTunnel是一个非常易于使用的超高性能分布式数据集成平台,支持海量数据的实时同步.它每天可以稳定高效地同步数百亿数据,并已用于近100家公司的生产. SeaTunne ...
数据集成框架SeaTunnel学习笔记
文章目录概述介绍应用场景插件支持情况安装和配置安装配置使用案例1入门案例2传参概述介绍 SeaTunnel 是一个简单易用的数据集成框架,在企业中,由于开发时间或开发部门不通用 ...
【SeaTunnel】从一个数据集成组件演化成企业级的服务
点亮 ⭐️ Star · 照亮开源之路 GitHub:https://github.com/apache/incubator-seatunnel 在 7 月 24 日 Apache SeaTunnel ...
盘点2021年晋升为Apache TLP的大数据相关项目
时间过得真快,2021年就过去了,又到了一年总结的时候了.本文将延续之前的惯例来总结一下过去一年大数据相关的项目顺利毕业成 Apache 顶级项目.在2021年一共有四个大数据相关项目顺利毕业成顶级项 ...
Shlle脚本传参调用seatunnel(原waterdrop)将hive中数据导入ClickHouse
前言公司分析数据已经存入hive,但需要输入参数计算得到很长一段时间的趋势变化数据(不固定查询),经调研ClickHouse时序优化后比较满足需求,并且ClickHouse在数据量大时最好采用DNS ...
【大数据】什么是数据集成？（SeaTunnel 集成工具介绍）
文章目录一.什么是数据集成? 二.ETL 又是什么? 三.SeaTunnel 介绍 1)概述 2)SeaTunnel 的作用 3)SeaTunnel 的特点 4)Seatunnel 优势与缺点 5) ...
seatunnel 高性能分布式数据集成平台
seatunnel 高性能分布式数据集成平台一.介绍二.为什么我们需要 seatunnel 三.seatunnel 使用场景四.seatunnel 的特性五.seatunnel 的工作流程六 ...
马蜂窝毕博：分析完这9点工作原理，我们最终选择了 Apache SeaTunnel！
点亮 ⭐️ Star · 照亮开源之路 https://github.com/apache/incubator-seatunnel 讲师简介毕博马蜂窝数据工程师在10月15日,Apac ...

seatunnel 简单使用(原名waterdrop)

业务背景

简介

安装

配置

插件开发

启动

seatunnel 简单使用(原名waterdrop)相关推荐

最新文章

热门文章