下面有些命令由于Kafka版本原因可能会报下面这种错误，请注意：

Apache Kafka: bootstrap-server is not a recognized option

2.5.0之前的版本使用  broker-list

2.5.0之后的版本使用  bootstrap-server

# Kafka启动服务

  1. 安装配置好zookeeper，添加好环境变量，打开cmd，输入命令启动服务。

zkServer

或

bin/zkServer.sh start

zkServer.sh stop

  1. 在%KAFKA_HOME%目录，按shift+鼠标右键，选择“在此处打开命令窗口”，在控制台输入命令启动服务。

.\bin\windows\kafka-server-start.bat .\config\server.properties

或

bin/kafka-server-start.sh config/server.properties &

bin/kafka-server-stop.sh config/server.properties &

kafka命令

  1. 创建主题

.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic flume

或

 bin/kafka-topics.sh --create --topic kafkatest2 --bootstrap-server 192.168.174.131:9092

  1. 查看主题

.\bin\windows\kafka-topics.bat --zookeeper localhost:2181 --list

  1. 启动生产者producer：

.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic 主题名

或

./bin/kafka-console-producer.sh --broker-list 172.17.0.103:9092 --topic flumeTopic

  1. 从文件中向主题批量发送数据：

.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic 主题名 < 文件名

或

 bin/kafka-console-producer.sh --topic testtopic01 --bootstrap-server 192.168.174.129:9092

  1. 启动消费者consumer

.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:2181 --topic 主题名 --from-beginning

或

bin/kafka-console-consumer.sh --topic flumeTopic --from-beginning --bootstrap-server 172.17.0.103:9092

此处遇到个问题：这样启动完成后zookeeper和consumer疯狂报错，consumer接收不到消息，且zookeeper显示 

后来查了一个国外大熊弟的解决办法，启动命令改为：菜鸟也不知道啥意思

.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic flume --from-beginning

以上部分关于Kafka的安装集群等详细教程可参考下面链接

https://blog.csdn.net/xiaolegeaizy/article/details/108723859 

# flume服务启动

bin/flume-ng agent --conf conf --conf-file conf/a1.conf --name a1 -Dflume.root.logger=INFO,console

新增a1.conf配置内容

  
# 定义 agent
a1.sources = src1
a1.channels = ch1
a1.sinks = k1
# 定义 sources
a1.sources.src1.type = exec
a1.sources.src1.command=tail -f /soft/flume/logs/log.txt
a1.sources.src1.channels=ch1
# 定义 sinks
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.topic = flumeTopic
a1.sinks.k1.brokerList = 172.17.0.103:9092
a1.sinks.k1.batchSize = 20
a1.sinks.k1.requiredAcks = 1
a1.sinks.k1.channel = ch1
# 定义 channels
a1.channels.ch1.type = memory
a1.channels.ch1.capacity = 1000

到此flume+Kafka简单集成完毕 。

# flink服务程序开发

flink配置详解

安装目录下主要有 flink-conf.yaml 配置、日志的配置文件、zk 配置、Flink SQL Client 配置。

基础配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# jobManager 的IP地址
jobmanager.rpc.address: localhost

# JobManager 的端口号
jobmanager.rpc.port: 6123

# JobManager JVM heap 内存大小
jobmanager.heap.size: 1024m

# TaskManager JVM heap 内存大小
taskmanager.heap.size: 1024m

# 每个 TaskManager 提供的任务 slots 数量大小

taskmanager.numberOfTaskSlots: 1

# 程序默认并行计算的个数
parallelism.default: 1

# 文件系统来源
# fs.default-scheme

高可用性配置

1
2
3
4
5
6
7
8
9
10
11

# 可以选择 'NONE' 或者 'zookeeper'.
# high-availability: zookeeper

# 文件系统路径，让 Flink 在高可用性设置中持久保存元数据
# high-availability.storageDir: hdfs:///flink/ha/

# zookeeper 集群中仲裁者的机器 ip 和 port 端口号
# high-availability.zookeeper.quorum: localhost:2181

# 默认是 open，如果 zookeeper security 启用了该值会更改成 creator
# high-availability.zookeeper.client.acl: open

容错和检查点 配置

1
2
3
4
5
6
7
8
9
10
11

# 用于存储和检查点状态
# state.backend: filesystem

# 存储检查点的数据文件和元数据的默认目录
# state.checkpoints.dir: hdfs://namenode-host:port/flink-checkpoints

# savepoints 的默认目标目录(可选)
# state.savepoints.dir: hdfs://namenode-host:port/flink-checkpoints

# 用于启用/禁用增量 checkpoints 的标志
# state.backend.incremental: false

web 前端配置

1
2
3
4
5
6
7
8

# 基于 Web 的运行时监视器侦听的地址.
#jobmanager.web.address: 0.0.0.0

#  Web 的运行时监视器端口
rest.port: 8081

# 是否从基于 Web 的 jobmanager 启用作业提交
# jobmanager.web.submit.enable: false

高级配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# io.tmp.dirs: /tmp

# 是否应在 TaskManager 启动时预先分配 TaskManager 管理的内存
# taskmanager.memory.preallocate: false

# 类加载解析顺序，是先检查用户代码 jar（“child-first”）还是应用程序类路径（“parent-first”）。 默认设置指示首先从用户代码 jar 加载类
# classloader.resolve-order: child-first


# 用于网络缓冲区的 JVM 内存的分数。 这决定了 TaskManager 可以同时拥有多少流数据交换通道以及通道缓冲的程度。 如果作业被拒绝或者您收到系统没有足够缓冲区的警告，请增加此值或下面的最小/最大值。 另请注意，“taskmanager.network.memory.min”和“taskmanager.network.memory.max”可能会覆盖此分数

# taskmanager.network.memory.fraction: 0.1
# taskmanager.network.memory.min: 67108864
# taskmanager.network.memory.max: 1073741824

1
2
3
4
5
6
7
8
9
10
11

# 指示是否从 Kerberos ticket 缓存中读取
# security.kerberos.login.use-ticket-cache: true

# 包含用户凭据的 Kerberos 密钥表文件的绝对路径
# security.kerberos.login.keytab: /path/to/kerberos/keytab

# 与 keytab 关联的 Kerberos 主体名称
# security.kerberos.login.principal: flink-user

# 以逗号分隔的登录上下文列表，用于提供 Kerberos 凭据（例如，`Client，KafkaClient`使用凭证进行 ZooKeeper 身份验证和 Kafka 身份验证）
# security.kerberos.login.contexts: Client,KafkaClient

Zookeeper 安全配置

1
2
3
4
5

# 覆盖以下配置以提供自定义 ZK 服务名称
# zookeeper.sasl.service-name: zookeeper

# 该配置必须匹配 "security.kerberos.login.contexts" 中的列表（含有一个）
# zookeeper.sasl.login-context-name: Client

HistoryServer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 你可以通过 bin/historyserver.sh (start|stop) 命令启动和关闭 HistoryServer

# 将已完成的作业上传到的目录
# jobmanager.archive.fs.dir: hdfs:///completed-jobs/

# 基于 Web 的 HistoryServer 的地址
# historyserver.web.address: 0.0.0.0

# 基于 Web 的 HistoryServer 的端口号
# historyserver.web.port: 8082

# 以逗号分隔的目录列表，用于监视已完成的作业
# historyserver.archive.fs.dir: hdfs:///completed-jobs/

# 刷新受监控目录的时间间隔（以毫秒为单位）
# historyserver.archive.fs.refresh-interval: 10000

查看下另外两个配置 slaves / master

2、slaves

里面是每个 worker 节点的 IP/Hostname，每一个 worker 结点之后都会运行一个 TaskManager，一个一行。

1

localhost

3、masters

host:port

1

localhost:8081

4、zoo.cfg

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

# 每个 tick 的毫秒数
tickTime=2000

# 初始同步阶段可以采用的 tick 数
initLimit=10

# 在发送请求和获取确认之间可以传递的 tick 数
syncLimit=5

# 存储快照的目录
# dataDir=/tmp/zookeeper

# 客户端将连接的端口
clientPort=2181

# ZooKeeper quorum peers
server.1=localhost:2888:3888
# server.2=host:peer-port:leader-port

5、日志配置

Flink 在不同平台下运行的日志文件

1
2
3
4
5
6
7

log4j-cli.properties
log4j-console.properties
log4j-yarn-session.properties
log4j.properties
logback-console.xml
logback-yarn.xml
logback.xml

sql-client-defaults.yaml

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

execution:
  # 'batch' or 'streaming' execution
  type: streaming
  # allow 'event-time' or only 'processing-time' in sources
  time-characteristic: event-time
  # interval in ms for emitting periodic watermarks
  periodic-watermarks-interval: 200
  # 'changelog' or 'table' presentation of results
  result-mode: changelog
  # parallelism of the program
  parallelism: 1
  # maximum parallelism
  max-parallelism: 128
  # minimum idle state retention in ms
  min-idle-state-retention: 0
  # maximum idle state retention in ms
  max-idle-state-retention: 0
  
deployment:
  # general cluster communication timeout in ms
  response-timeout: 5000
  # (optional) address from cluster to gateway
  gateway-address: ""
  # (optional) port from cluster to gateway
  gateway-port: 0

Flink sql client ：你可以从官网这里了解 https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/sqlClient.html

flinke启动：

 bin/start-cluster.sh

jar包程序执行（flink程序开发）

./bin/flink run --class org.example.BatchJob ./examples/cs-1.0-SNAPSHOT.jar 

# flume # kafka # flink
Logo
Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

Kafka入门（一） 概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

avatar Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

avatar Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...

avatar Kafka开源项目指南