flink用的binlog监听source

发布 : 2019-04-24 分类 : 工具类 浏览 :

flink用的influxdb用sink

前言

会有一系列的文章介绍common-*.jar的各种用法,这些工具类jar包都已上传在maven中央库。可以直接通过maven坐标引入使用。源码可以参见:https://gitee.com/rjzjh/common

场景分析

flink的stream模式有时间窗口,它会把一段时间的指标值不间断的输出,它是以时间为横坐标,且是无限的,这是典型的时序数据库的应用场景。但Flink默认并没有influxdb的Sink输出。common-flink-sink-influxdb.jar就是为满足此场景下而开发的。

使用示例

使用前需要引用依赖包,maven坐标为:

1
2
3
4
5
<dependency>
<groupId>net.wicp.tams</groupId>
<artifactId>common-flink-sink-influxdb</artifactId>
<version>3.5.15</version>
</dependency>

调用示例方式:

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
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import net.wicp.tams.common.flink.sink.influxdb.InfluxDBConfig;
import net.wicp.tams.common.flink.sink.influxdb.InfluxDBPoint;
import net.wicp.tams.common.flink.sink.influxdb.InfluxDBSink;

public class InfluxDBSinkExample {
private static final int N = 10000;

public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

List<String> dataList = new ArrayList<>();
for (int i = 0; i < N; ++i) {
String id = "server" + String.valueOf(i);
dataList.add("cpu#" + id);
dataList.add("mem#" + id);
dataList.add("disk#" + id);
}
DataStream<String> source = env.fromElements(dataList.toArray(new String[0]));

DataStream<InfluxDBPoint> dataStream = source.map(new RichMapFunction<String, InfluxDBPoint>() {
@Override
public InfluxDBPoint map(String s) throws Exception {
String[] input = s.split("#");

String measurement = input[0];
long timestamp = System.currentTimeMillis();

HashMap<String, String> tags = new HashMap<>();
tags.put("host", input[1]);
tags.put("region", "region#" + String.valueOf(input[1].hashCode() % 20));

HashMap<String, Object> fields = new HashMap<>();
fields.put("value1", input[1].hashCode() % 100);
fields.put("value2", input[1].hashCode() % 50);

return new InfluxDBPoint(measurement, timestamp, tags, fields);
}
});

// dataStream.print();
InfluxDBConfig influxDBConfig = InfluxDBConfig.builder("http://localhost:8086", "db_flink_test")
.batchActions(1000).flushDuration(100, TimeUnit.MILLISECONDS).createDatabase(true).enableGzip(true)
.build();

dataStream.addSink(new InfluxDBSink(influxDBConfig));

env.execute("InfluxDB Sink Example");
}

}
本文作者 : andy.zhou
原文链接 : https://rjzjh.gitee.io/2019/04/24/common-influxdb/
版权声明 : 本博客所有文章除特别声明外,均采用 CC BY-NC-SA 4.0 许可协议。转载请注明出处!

知识 & 情怀 | 二者兼得

微信扫一扫, 向我投食

微信扫一扫, 向我投食

支付宝扫一扫, 向我投食

支付宝扫一扫, 向我投食

留下足迹