Flink 系列文章

一、Flink 专栏

Flink 专栏系统介绍某一知识点，并辅以具体的示例进行说明。

• 1、Flink 部署系列
  本部分介绍Flink的部署、配置相关基础内容。

• 2、Flink基础系列
  本部分介绍Flink 的基础部分，比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。

• 3、Flik Table API和SQL基础系列
  本部分介绍Flink Table Api和SQL的基本用法，比如Table API和SQL创建库、表用法、查询、窗口函数、catalog等等内容。

• 4、Flik Table API和SQL提高与应用系列
  本部分是table api 和sql的应用部分，和实际的生产应用联系更为密切，以及有一定开发难度的内容。

• 5、Flink 监控系列
  本部分和实际的运维、监控工作相关。

二、Flink 示例专栏

Flink 示例专栏是 Flink 专栏的辅助说明，一般不会介绍知识点的信息，更多的是提供一个一个可以具体使用的示例。本专栏不再分目录，通过链接即可看出介绍的内容。

两专栏的所有文章入口点击：Flink 系列文章汇总索引

---

---

本文主要介绍Flink 的clickhouse作为数据源的使用，以一个简单的示例进行展示。

如果需要了解更多内容，可以在本人Flink 专栏中了解更新系统的内容。

本文除了maven依赖外，没有其他依赖。

本文依赖clickhouse数据库环境好用。

本专题分为以下几篇文章：
【flink番外篇】3、fflink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例（1） - File、Socket、Collection
【flink番外篇】3、fflink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例（2）- 自定义、mysql
【flink番外篇】3、flink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例（3）- kafka
【flink番外篇】3、flink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例（4）- redis -异步读取
【flink番外篇】3、flink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例（5）- clickhouse
【flink番外篇】3、flink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例 - 完整版

一、maven依赖

本文依赖见【flink番外篇】3、flink的source介绍及示例（1）- File、Socket、Collection，不再赘述。

如果有新增的maven依赖，则会在示例时加以说明，避免篇幅的过大。

二、关于clickhouse的介绍

关于clickhouse的基本知识详见该系列文章，关于该部分不再赘述。

ClickHouse系列文章
1、ClickHouse介绍
2、clickhouse安装与简单验证（centos）
3、ClickHouse表引擎-MergeTree引擎
4、clickhouse的Log系列表引擎、外部集成表引擎和其他特殊的表引擎介绍及使用
5、ClickHouse查看数据库容量、表的指标、表分区、数据大小等

三、clickhouse作为Flink的source示例

1、maven依赖

<dependency><groupId>ru.yandex.clickhouse</groupId><artifactId>clickhouse-jdbc</artifactId><version>0.1.40</version>
</dependency>

2、实现

1）、user bean

import lombok.Data;/*** @author alanchan**/
@Data
public class UserSource {private int id;private String name;private int age;public UserSource(int id, String name, int age) {this.id = id;this.name = name;this.age = age;}public UserSource() {}
}

2）、source实现

import java.sql.ResultSet;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;import ru.yandex.clickhouse.ClickHouseConnection;
import ru.yandex.clickhouse.ClickHouseDataSource;
import ru.yandex.clickhouse.ClickHouseStatement;
import ru.yandex.clickhouse.settings.ClickHouseProperties;
import ru.yandex.clickhouse.settings.ClickHouseQueryParam;/*** @author alanchan**/
public class Source_Clickhouse {/*** @param args* @throws Exception*/public static void main(String[] args) throws Exception {// envStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);// sourceDataStream<UserSource> users = env.addSource(new ClickhouseSource());// transformation// sinkusers.print();// executeenv.execute();}private static class ClickhouseSource extends RichParallelSourceFunction<UserSource> {private boolean flag = true;private ClickHouseConnection conn = null;private ClickHouseStatement stmt = null;private ResultSet rs = null;private Map<ClickHouseQueryParam, String> additionalDBParams = new HashMap<>();UserSource user = null;private String sql = "select id,name,age from t_flink_sink_clickhouse";// open只执行一次,适合开启资源@Overridepublic void open(Configuration parameters) throws Exception {ClickHouseProperties properties = new ClickHouseProperties();String url = "jdbc:clickhouse://192.168.10.42:8123/tutorial";properties.setSessionId(UUID.randomUUID().toString());
//			properties.setDatabase("tutorial");
//			properties.setHost("192.168.10.42");ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);// ClickHousePropertiesadditionalDBParams.put(ClickHouseQueryParam.SESSION_ID, UUID.randomUUID().toString());conn = dataSource.getConnection();stmt = conn.createStatement();}@Overridepublic void run(SourceContext<UserSource> ctx) throws Exception {while (flag) {rs = stmt.executeQuery(sql, additionalDBParams);while (rs.next()) {user = new UserSource(rs.getInt(1), rs.getString(2), rs.getInt(3));ctx.collect(user);}}}// 接收到cancel命令时取消数据生成@Overridepublic void cancel() {flag = false;}@Overridepublic void close() throws Exception {if (conn != null)conn.close();if (stmt != null)stmt.close();if (rs != null)rs.close();}}
}

3、验证

启动应用程序，查看应用程序控制台输出是不是与期望的一致即可。

1）、clickhouse数据源准备

1、启动clickhouse服务
2、创建数据库 tutorial
3、创建表 t_flink_sink_clickhouse
4、插入数据并查询

server2 :) select * from t_flink_sink_clickhouse limit 10;SELECT *
FROM t_flink_sink_clickhouse
LIMIT 10Query id: cd8bbe95-e8b0-448d-a7e2-dae3b5b2602d┌─id─┬─name─────┬─age─┐
│  1 │ alanchan │  19 │
│  2 │ alan     │  20 │
│  3 │ chan     │  21 │
└────┴──────────┴─────┘3 rows in set. Elapsed: 0.052 sec. 

2）、启动应用程序并观察控制台输出

由于本应用程序未设置查询频率，所以会一直输出，可以设置以一定的频率来查询。

4> UserSource(id=1, name=alanchan, age=19)
4> UserSource(id=2, name=alan, age=20)
4> UserSource(id=3, name=chan, age=21)

以上，本文主要介绍Flink 的clickhouse作为数据源的使用，以一个简单的示例进行展示。

如果需要了解更多内容，可以在本人Flink 专栏中了解更新系统的内容。

本专题分为以下几篇文章：
【flink番外篇】3、fflink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例（1） - File、Socket、Collection
【flink番外篇】3、fflink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例（2）- 自定义、mysql
【flink番外篇】3、flink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例（3）- kafka
【flink番外篇】3、flink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例（4）- redis -异步读取
【flink番外篇】3、flink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例（5）- clickhouse
【flink番外篇】3、flink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例 - 完整版