Nifi + Kafka + mysql构建实时数仓_nifi串接kafka

标题Nifi + Kafka + mysql构建实时数仓

Nifi 安装 Nifi安装
Kafka安装 kafka安装

一 Nifi 简介

Nifi 最重要的就是Processor（数据处理单元）和FlowFlie（数据流）

• FlowFile: 每条“用户数据”（即，用户通过NiFi获取或者是生成的，需要进行处理和分发的数据）均被称为FlowFile。 FlowFile由两部分组成：属性和内容。 内容是用户数据本身， 属性是与用户数据关联的键值对，用来提供一些额外的信息。比如fliename，表示的就是FlowFile的文件名,如果是通过GetFile从操作系统中取的文件的话,则filename就表示的是取到的文件的名称.

  Processor：是NiFi中负责创建、发送、接收、转换、路由、拆分、合并和处理FlowFiles的业务组件。 它是NiFi用户可用于构建其数据流的最重要的构建块。

二 Nifi简单使用 移动文件

拖动图中红色创建一个组，然后在组中进行数据处理


工程完整数据流

分步骤创建
1.创建getfile

配置getfile

填写要复制文件所在的目录

同理创建一个putfile


身材flowfile

enable Getfile
然后start

同理putfile也要enable，然后start

三 Nifi实时 数仓 将生产系统mysql数据实时导入kafka 进行数据分析

Nifi 实时数仓建设全流程，数据推到kafka后，应该用flink进行实时计算和处理，这里先不介绍如何利用flink进行实时处理。只把数据推到kafka。

3.1 表连接

3.2 定时生成sql

创建jsonrecordsetwriter





naespace :数据库名
info_test 表名

{
	"namespace": "数据库名",
	"type": "record",
	"name": "info_test",
	"fields": [
		{
			"name": "id",
			"type": [
				"int",
				"null"
			]
		},
		{
			"name": "store_name",
			"type": [
				"string",
				"null"
			]
		},
		{
			"name": "store_code",
			"type": [
				"string",
				"null"
			]
		},
		{
			"name": "product_name",
			"type": [
				"string",
				"null"
			]
		},
		{
			"name": "src_tbName",
			"type": "string",
			"default": "info_test"
		}
	]
}
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

然后点击生效

3.3 分割json

配置很简单

3.4 向kafka发送消息

全部启动后，就能在kafka拿到消息