广告设计主要学什么内容_自建小程序商城_上海app网络推广公司电话_小说关键词搜索器

1、hadoop集群连接本地MySQL

1.1 首先测试集群是否可以ping通本地

虚拟机可以ping通网关（192.168.10.2），但不能ping通192.168.10.1，这表明问题可能出在Windows主机的防火墙设置或VMware的网络配置上。

1.1.1 检查Windows防火墙设置

确保Windows防火墙没有阻止ICMP请求（ping）可以临时禁用Windows防火墙进行测试：

打开命令提示符（以管理员身份运行）
输入以下命令来禁用防火墙

netsh advfirewall set allprofiles state off

然后再次尝试从虚拟机ping 192.168.10.1。如果现在可以ping通，说明防火墙是问题所在。需要添加一个允许ICMP的规则。

1.1.2 重新启用防火墙并添加规则

输入以下命令来重新启用防火墙

netsh advfirewall set allprofiles state on

在命令提示符中输入以下命令来添加允许ICMP请求的规则

netsh advfirewall firewall add rule name="Allow ICMPv4" protocol=icmpv4:8,any dir=in action=allow

在命令提示符中输入以下命令来添加允许3306端口的入站规则

netsh advfirewall firewall add rule name="Allow MySQL 3306" dir=in action=allow protocol=TCP localport=3306

1.2 虚拟机连接到物理机上的MySQL服务器

需要确保几个关键点：

MySQL服务器配置：确保MySQL服务器允许远程连接。
网络配置：确保虚拟机和物理机在同一网络中，并且可以相互通信。
防火墙设置：确保没有防火墙阻止MySQL的端口（默认是3306）。
MySQL用户权限：确保有用户权限从虚拟机的IP地址连接到MySQL服务器。

1.2.1 配置MySQL服务器以允许远程连接

登录MySQL服务器
```
mysql -uroot -p123456
```
查看当前的用户权限
```
SHOW GRANTS FOR 'root'@'localhost';
```

创建一个允许远程连接的用户

CREATE USER 'remote_user'@'%' IDENTIFIED BY '123456';

给这个用户授权：

GRANT ALL PRIVILEGES ON *.* TO 'remote_user'@'%' WITH GRANT OPTION;
FLUSH PRIVILEGES;

确保MySQL配置文件（my.cnf或my.ini）允许远程连接：
- 找到MySQL的配置文件，通常在/etc/mysql/目录下。
- 确保bind-address设置为0.0.0.0或者被注释掉，这样MySQL服务器可以监听所有网络接口。

在虚拟机上执行如下命令：

mysql -h 192.168.10.1 -u remote_user -p

已经成功配置了从Hadoop集群的hadoop102节点访问本地MySQL服务器，并且已经解决了网络和防火墙的问题，现在可以使用SeaTunnel来全量同步news_data数据库的数据到HDFS上。

2、seatunnel导入数据库全量数据到HDFS（单表）

2.1 下载并安装SeaTunnel

官方网站：Apache SeaTunnel | Apache SeaTunnel

导入到/opt/software下，执行如下命令解压

tar -xzvf apache-seatunnel-2.3.8-bin.tar.gz -C /opt/module/

2.2 安装Connector插件

从2.2.0-beta版本开始，二进制包默认不提供connector依赖。因此，需要执行以下命令来安装所需的connector插件：

sh bin/install_plugin.sh $version

2.3 驱动jar包

下载MySQL JDBC驱动：确保下载了正确版本的MySQL JDBC驱动jar包。从MySQL官网下载。
将驱动添加到SeaTunnel的lib目录：将下载的MySQL JDBC驱动jar包放到SeaTunnel的lib目录下。这样，当SeaTunnel启动时，它就能够加载所需的MySQL驱动类

2.4 配置SeaTunnel任务

配置文件mysql_to_hdfs.conf，并将其放置在SeaTunnel的config目录下。这个配置文件应该包含了源（MySQL）和接收器（HDFS）的正确配置。

env {execution.parallelism = 1 # 根据你的集群资源调整并行度
}source {JDBC {driver = "com.mysql.cj.jdbc.Driver"url = "jdbc:mysql://192.168.10.1:3306/news_data?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC"user = "remote_user"password = "123456"query = "SELECT * FROM news_information"  # 使用查询语句来导入数据}
}sink {HdfsFile {fs.defaultFS = "hdfs://hadoop102:8020"  # HDFS NameNode 地址path = "/news"  # HDFS 目标目录路径file_type = "text"  # 文件类型，例如 text, csv, orc 等field_delimiter = ","  # 字段分隔符，根据实际情况调整file_format_type = "text"  # 指定文件格式类型为文本}
}

选择执行模式

SeaTunnel支持多种执行模式，包括本地模式（local）、独立模式（standalone）、集群模式（yarn/cluster）等。对于大多数大数据任务，推荐使用集群模式来充分利用Hadoop集群的资源。

2.5 环境变量配置

/opt/module/hadoop/etc/hadoop是包含Hadoop配置文件的目录。需要将这个目录路径设置为HADOOP_CONF_DIR环境变量，以便SeaTunnel的Spark作业能够正确地连接到YARN集群。

使用如下命令

sudo vim /etc/profile.d/my_env.sh

在环境变量最后添加如下

export HADOOP_CONF_DIR=/opt/module/hadoop/etc/hadoop

运行以下命令使变量立即生效

source /etc/profile.d/my_env.sh

运行命令

./bin/start-seatunnel-spark-3-connector-v2.sh --config ./config/mysql_to_hdfs.conf --master yarn

SeaTunnel作业通过Spark提交到YARN集群，启动一个SeaTunnel作业，从MySQL读取数据并将其写入到HDFS中，使用YARN作为资源管理器。

3、seatunnel导入数据库全量数据到HDFS（多表）

在seatunnel的安装目录下创建脚本mysql_full_hdfs.sh

#!/bin/bash# 获取当前日期
current_date=$(date +%Y-%m-%d)# 定义数据库连接信息
DB_URL="jdbc:mysql://192.168.10.1:3306/test?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC"
DB_USER="remote_user"
DB_PASSWORD="123456"# 定义HDFS路径
HDFS_PATH="hdfs://hadoop102:8020"# 定义表名数组
TABLES=("stu" "t_user")# 循环处理每个表
for TABLE in "${TABLES[@]}"; do# 创建配置文件内容，包含当前日期conf_file="/opt/module/apache-seatunnel-2.3.8/config/${TABLE}_to_hdfs_${current_date}.conf"cat > "$conf_file" << EOF
env {execution.parallelism = 1
}
source {JDBC {driver = "com.mysql.cj.jdbc.Driver"url = "${DB_URL}"user = "${DB_USER}"password = "${DB_PASSWORD}"query = "SELECT * FROM ${TABLE}"}
}
sink {HdfsFile {fs.defaultFS = "${HDFS_PATH}"path = "/news/${TABLE}_full/${current_date}"  # HDFS 目标目录路径，每个表都有自己的目录，并且包含日期file_type = "parquet"  # 文件类型改为Parquetfile_format_type = "parquet"  # 文件格式类型为Parquetparquet_compression_CODEC = "gzip"  # 使用gzip压缩}
}
EOF# 运行Seatunnel任务/opt/module/apache-seatunnel-2.3.8/bin/start-seatunnel-spark-3-connector-v2.sh --config "$conf_file" --master yarn# 删除配置文件rm "$conf_file"
done

使用chmod +x 脚本，再运行脚本。

4、MySQL到Kafka的实时数据流和变更数据捕获（单节点）

Debezium在Kafka Connect集群中监控MySQL数据库变更

4.1 MySQL配置

创建用户：首先，需要使用 CREATE USER 命令来创建一个新的用户。

CREATE USER 'debezium'@'192.168.10.102' IDENTIFIED BY '123456';

授予权限：然后，可以使用 GRANT 命令来授予这个用户所需的权限。

GRANT SUPER, REPLICATION CLIENT, SELECT ON *.* TO 'debezium'@'192.168.10.102';

为用户 debezium 授予 LOCK TABLES 权限

请注意，授予 LOCK TABLES 权限可能会影响数据库的并发性，因为它允许用户锁定表。在授予权限时，应始终确保遵循最小权限原则，只授予用户完成其任务所需的权限。

GRANT SUPER, REPLICATION CLIENT, SELECT ON *.* TO 'debezium'@'192.168.10.102';

为用户 debezium 授予 RELOAD 权限

GRANT RELOAD ON *.* TO 'debezium'@'192.168.10.102';

请注意，授予 RELOAD 权限允许用户执行 FLUSH TABLES WITH READ LOCK 命令，这在 Debezium 连接器进行初始快照时是必要的。确保在授予权限时，用户 debezium 已经存在于 MySQL 中，并且使用的是正确的主机地址（在本例中为虚拟机的 IP 地址 192.168.10.102）。

刷新权限：最后，需要刷新 MySQL 的权限设置，使新权限生效：

FLUSH PRIVILEGES;

请注意，上述步骤中的 '192.168.10.102' 应该替换为虚拟机的 IP 地址，因为这是 Kafka Connect 尝试连接到 MySQL 数据库的地址。

MySQL的配置文件：

# 后续换为允许的IP
bind-address = 0.0.0.0
log-bin=mysql-bin
binlog-format = ROW
binlog-row-image = FULL
expire_logs_days = 10

4.2 下载安装Zookeeper

在zoo.cfg下配置如下信息：

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/opt/module/zookeeper/zkData
clientPort=2181

在bin目录下执行

./zkServer.sh start

4.3 下载安装Kafka

在config目录的server.properties配置文件中，配置如下信息：

broker.id=0
advertised.listeners=PLAINTEXT://hadoop102:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/module/kafka/datas
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.retention.check.interval.ms=300000
zookeeper.connect=hadoop102:2181/kafka
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

在bin目录下执行

./kafka-server-start.sh /opt/module/kafka/config/server.properties

4.4 下载安装Debezium

下载到/opt/software

 wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/1.9.5.Final/debezium-connector-mysql-1.9.5.Final-plugin.tar.gz

解压到：/opt/module/kafka/libs 下

tar -xzf debezium-connector-mysql-1.9.5.Final-plugin.tar.gz -C /opt/module/kafka/libs/

把 debezium-connector-mysql 里面的jar包移动到libs目录下。

cp debezium-connector-mysql/*.jar /opt/module/kafka/libs/

在/opt/module/kafka/config目录下的connect-distributed.properties 配置文件中添加

plugin.path=/opt/module/kafka/libs

在config目录下创建配置文件mysql-connector.json，并且添加如下信息

{"name":"mysql-connector","config":{"connector.class":"io.debezium.connector.mysql.MySqlConnector","tasks.max":"1","database.hostname":"192.168.10.1","database.port":"3306","database.user":"debezium","database.password":"123456","database.server.id":"184054","database.server.name":"mydb","database.include.list":"test","database.history.kafka.bootstrap.servers":"hadoop102:9092","schema.history.internal.kafka.bootstrap.servers":"hadoop102:9092","schema.history.internal.kafka.topic":"schema-changes.test","key.converter":"org.apache.kafka.connect.json.JsonConverter","key.converter.schemas.enable":"false","value.converter":"org.apache.kafka.connect.json.JsonConverter","value.converter.schemas.enable":"false","database.history.kafka.topic":"dbhistory.test","database.ssl.mode":"REQUIRED","database.ssl.truststore.location":"/path/to/truststore.jks","database.ssl.truststore.password":"truststore-password","database.connectionTimeZone":"UTC"}}

在config目录下创建配置文件debezium.properties,如下信息：

bootstrap.servers=hadoop102:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false
offset.storage.topic=connect-offsets
offset.flush.interval.ms=10000
# 设置 offset topic 的复制因子为 1
config.storage.topic=connect-configs
config.storage.replication.factor=1
offset.storage.replication.factor=1
status.storage.topic=connect-status
status.storage.replication.factor=1
status.flush.interval.ms=5000
group.id=connect-cluster
name=mysql-connector
connector.class=io.debezium.connector.mysql.MySqlConnector
tasks.max=1
database.hostname=192.168.10.1
database.port=3306
database.user=debezium
database.password=123456
database.server.id=184054
database.server.name=mydb
database.history.kafka.bootstrap.servers=hadoop102:9092
database.history.kafka.topic=dbhistory.mydb
database.history.store.only.monitored.tables.ddl=true
table.include.list=test.stu,test.t_user

将zookeeper和kafka启动起来，再启动kafka-connect。

kafka的bin目录下

./connect-distributed.sh /opt/module/kafka/config/debezium.properties

执行注册命令，kafka的config目录下

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://hadoop102:8083/connectors/ -d @mysql-connector.json

查看状态

curl -s http://hadoop102:8083/connectors/mysql-connector/status

删除

url -X DELETE localhost:8083/connectors/mysql-connector

5、MySQL到Kafka的实时数据流和变更数据捕获（多节点）

5.1 Zookeeper配置

hadoop102节点上

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/opt/module/zookeeper/zkData
clientPort=2181
server.2=hadoop102:2888:3888
server.3=hadoop103:2888:3888
server.4=hadoop104:2888:3888

/opt/module/zookeeper/zkData目录下有一个文件myid,里面的内容是：2

hadoop103节点上

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/opt/module/zookeeper/zkData
clientPort=2181
server.2=hadoop102:2888:3888
server.3=hadoop103:2888:3888
server.4=hadoop104:2888:3888

/opt/module/zookeeper/zkData目录下有一个文件myid,里面的内容是：3

hadoop104节点上

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/opt/module/zookeeper/zkData
clientPort=2181
server.2=hadoop102:2888:3888
server.3=hadoop103:2888:3888
server.4=hadoop104:2888:3888

/opt/module/zookeeper/zkData目录下有一个文件myid,里面的内容是：4

5.2 Kafka配置

hadoop102节点上

broker.id=0
advertised.listeners=PLAINTEXT://hadoop102:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/opt/module/kafka/datas
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.retention.check.interval.ms=300000
zookeeper.connect=hadoop102:2181,hadoop103:2181,hadoop104:2181/kafka
zookeeper.connection.timeout.ms=18000
group.initial.rebalance.delay.ms=0

hadoop103节点上

只需要修改kafka/config/server.properties

broker.id=1
advertised.listeners=PLAINTEXT://hadoop103:9092

hadoop104节点上