2. 数据转换策略：
– ETL（Extract, Transform, Load）：提取、转换、加载数据
– ELT（Extract, Load, Transform）：提取、加载、转换数据
– 数据清洗：处理数据质量问题

3. 数据存储策略：
– 分层存储：热数据存储在MySQL，冷数据存储在大数据系统
– 分区存储：根据时间或其他维度对数据进行分区
– 压缩存储：对冷数据进行压缩，减少存储空间

4. 查询策略：
– 联邦查询：在MySQL和大数据系统之间执行联合查询
– 数据缓存：缓存频繁查询的结果
– 索引优化：优化MySQL和大数据系统的索引

5. 安全策略：
– 数据加密：加密传输和存储的数据
– 访问控制：实施基于角色的访问控制
– 审计日志：记录数据访问和操作

6. 监控策略：
– 数据同步监控：监控数据同步的状态和性能
– 系统监控：监控MySQL和大数据系统的性能
– 告警机制：设置告警规则，及时发现问题

7. 灾备策略：
– 数据备份：定期备份MySQL和大数据系统的数据
– 灾难恢复：制定灾难恢复计划
– 高可用性：确保系统的高可用性

# 集成策略最佳实践
1. 明确业务需求：根据业务需求选择合适的集成策略
2. 数据建模：设计合理的数据模型，确保数据一致性
3. 性能优化：优化数据同步和查询性能
4. 安全加固：实施安全措施，保护数据安全
5. 监控完善：建立完善的监控系统，及时发现问题
6. 文档化：记录集成策略和流程，便于后续维护
7. 测试验证：在生产环境部署前进行充分的测试
8. 持续改进：根据实际情况持续改进集成策略

2. 大数据系统优化：
– 资源配置：根据数据量和处理需求配置足够的资源
– 数据分区：对数据进行合理分区，提高并行处理能力
– 缓存策略：使用缓存，减少重复计算
– 并行处理：利用大数据系统的并行处理能力

3. 数据传输优化：
– 批量传输：使用批量传输，减少网络开销
– 压缩传输：压缩数据，减少传输时间
– 增量同步：只同步变化的数据，减少数据传输量
– 网络优化：优化网络配置，提高传输速度

4. 数据存储优化：
– 数据压缩：对冷数据进行压缩，减少存储空间
– 存储格式：选择合适的存储格式，如Parquet、ORC等
– 分区策略：根据时间或其他维度对数据进行分区
– 存储介质：使用SSD等高性能存储介质

5. 处理优化：
– 批处理优化：优化批处理作业，提高处理效率
– 流处理优化：优化流处理作业，减少延迟
– 任务调度：合理调度任务，避免资源争用
– 负载均衡：实现负载均衡，提高系统利用率

6. 监控与调优：
– 性能监控：监控系统性能，发现瓶颈
– 日志分析：分析系统日志，发现问题
– 性能测试：定期进行性能测试，评估系统性能
– 持续调优：根据监控数据持续调优系统

# 性能优化最佳实践
1. 基准测试：在优化前进行基准测试，确定性能基线
2. 目标明确：明确性能优化的目标，如响应时间、吞吐量等
3. 瓶颈分析：找出系统的性能瓶颈，针对性地进行优化
4. 分步优化：分步进行优化，每次优化后进行测试
5. 监控反馈：建立监控系统，实时反馈优化效果
6. 文档记录：记录优化过程和结果，便于后续参考
7. 持续优化：根据业务需求和数据量的变化，持续优化系统
8. 团队协作：加强开发、运维和数据分析团队的协作，共同优化系统

# 步骤2：安装Sqoop
# 下载Sqoop
tar -xzf sqoop-1.4.7.bin__hadoop-2.6.0.tar.gz
mv sqoop-1.4.7.bin__hadoop-2.6.0 /opt/sqoop

# 配置Sqoop
export SQOOP_HOME=/opt/sqoop
export PATH=$PATH:$SQOOP_HOME/bin

# 步骤3：配置MySQL驱动
# 将MySQL驱动复制到Sqoop的lib目录
cp mysql-connector-java-8.0.29.jar /opt/sqoop/lib/

# 步骤4：从MySQL导入数据到HDFS
# 使用Sqoop从MySQL导入数据到HDFS
sqoop import \
–connect jdbc:mysql://localhost:3306/app_db \
–username root \
–password rootpassword \
–table users \
–target-dir /user/hadoop/users \
–m 1

# 步骤5：从MySQL导入数据到Hive
# 使用Sqoop从MySQL导入数据到Hive
sqoop import \
–connect jdbc:mysql://localhost:3306/app_db \
–username root \
–password rootpassword \
–table users \
–hive-import \
–hive-table app_db.users \
–m 1

# 步骤6：从HDFS导出数据到MySQL
# 使用Sqoop从HDFS导出数据到MySQL
sqoop export \
–connect jdbc:mysql://localhost:3306/app_db \
–username root \
–password rootpassword \
–table users \
–export-dir /user/hadoop/users \
–input-fields-terminated-by ‘\t’

# 步骤7：增量导入数据
# 使用Sqoop增量导入数据
sqoop import \
–connect jdbc:mysql://localhost:3306/app_db \
–username root \
–password rootpassword \
–table users \
–target-dir /user/hadoop/users \
–m 1 \
–incremental append \
–check-column id \
–last-value 100

# 步骤8：验证集成
# 查看HDFS中的数据
hdfs dfs -ls /user/hadoop/users
hdfs dfs -cat /user/hadoop/users/part-m-00000

# 查看Hive中的数据
hive -e “SELECT * FROM app_db.users LIMIT 10;”

# MySQL与Hadoop集成最佳实践
1. 合理配置Sqoop参数：根据数据量和网络情况配置合适的参数
2. 使用增量导入：只导入变化的数据，减少数据传输量
3. 并行处理：使用多个map任务并行导入数据，提高效率
4. 数据压缩：启用数据压缩，减少存储空间和传输时间
5. 错误处理：配置错误处理策略，确保数据导入的可靠性
6. 监控与日志：开启监控和日志，及时发现和处理问题
7. 测试验证：在生产环境部署前进行充分的测试
8. 性能优化：根据实际情况优化Sqoop和Hadoop的配置

# 配置Spark
export SPARK_HOME=/opt/spark
export PATH=$PATH:$SPARK_HOME/bin

# 步骤2：准备MySQL驱动
# 将MySQL驱动复制到Spark的jars目录
cp mysql-connector-java-8.0.29.jar /opt/spark/jars/

# 步骤3：使用Spark读取MySQL数据
# 创建Spark应用读取MySQL数据
cat << EOF > spark-mysql-read.py
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName(“MySQL Integration”).getOrCreate()

# 读取MySQL数据
df = spark.read.format(“jdbc”).option(“url”, “jdbc:mysql://localhost:3306/app_db”).option(“driver”, “com.mysql.cj.jdbc.Driver”).option(“dbtable”, “users”).option(“user”, “root”).option(“password”, “rootpassword”).load()

# 显示数据
df.show()

# 执行数据分析
df.groupBy(“department”).count().show()
EOF

# 运行Spark应用
spark-submit spark-mysql-read.py

# 步骤4：使用Spark写入MySQL数据
# 创建Spark应用写入MySQL数据
cat << EOF > spark-mysql-write.py
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType

spark = SparkSession.builder.appName(“MySQL Integration”).getOrCreate()

# 创建测试数据
data = [(1, “user1”, “IT”), (2, “user2”, “HR”), (3, “user3”, “Finance”)]
schema = StructType([StructField(“id”, IntegerType(), True), StructField(“name”, StringType(), True), StructField(“department”, StringType(), True)])
df = spark.createDataFrame(data, schema)

# 写入MySQL数据
df.write.format(“jdbc”).option(“url”, “jdbc:mysql://localhost:3306/app_db”).option(“driver”, “com.mysql.cj.jdbc.Driver”).option(“dbtable”, “users”).option(“user”, “root”).option(“password”, “rootpassword”).mode(“append”).save()
EOF

# 运行Spark应用
spark-submit spark-mysql-write.py

# 步骤5：使用Spark SQL分析MySQL数据
# 创建Spark应用使用Spark SQL分析MySQL数据
cat << EOF > spark-mysql-sql.py
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName(“MySQL Integration”).getOrCreate()

# 读取MySQL数据
df = spark.read.format(“jdbc”).option(“url”, “jdbc:mysql://localhost:3306/app_db”).option(“driver”, “com.mysql.cj.jdbc.Driver”).option(“dbtable”, “users”).option(“user”, “root”).option(“password”, “rootpassword”).load()

# 创建临时表
df.createOrReplaceTempView(“users”)

# 执行SQL查询
result = spark.sql(“SELECT department, COUNT(*) as count FROM users GROUP BY department”)
result.show()
EOF

# 运行Spark应用
spark-submit spark-mysql-sql.py

# 步骤6：验证集成
# 查看MySQL中的数据
mysql -u root -p -e “SELECT * FROM app_db.users;”

# MySQL与Spark集成最佳实践
1. 合理配置Spark参数：根据数据量和处理需求配置合适的参数
2. 使用分区读取：对大表使用分区读取，提高并行处理能力
3. 缓存数据：对频繁使用的数据进行缓存，提高查询性能
4. 批量操作：使用批量读写，减少数据库连接开销
5. 错误处理：配置错误处理策略，确保数据处理的可靠性
6. 监控与日志：开启监控和日志，及时发现和处理问题
7. 测试验证：在生产环境部署前进行充分的测试
8. 性能优化：根据实际情况优化Spark和MySQL的配置

# 配置Kafka
export KAFKA_HOME=/opt/kafka
export PATH=$PATH:$KAFKA_HOME/bin

# 步骤2：启动Kafka服务
# 启动Zookeeper
$KAFKA_HOME/bin/zookeeper-server-start.sh -daemon $KAFKA_HOME/config/zookeeper.properties

# 启动Kafka broker
$KAFKA_HOME/bin/kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

# 步骤3：创建Kafka主题
# 创建主题
$KAFKA_HOME/bin/kafka-topics.sh –create –topic mysql-data –bootstrap-server localhost:9092 –partitions 1 –replication-factor 1

# 步骤4：安装Debezium
# 下载Debezium connector
wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/1.8.1.Final/debezium-connector-mysql-1.8.1.Final-plugin.tar.gz

# 解压到Kafka connect插件目录
tar -xzf debezium-connector-mysql-1.8.1.Final-plugin.tar.gz -C /opt/kafka/plugins/

# 步骤5：配置MySQL
# 启用binlog
# 编辑my.cnf文件
[mysqld]
server-id = 1
log-bin = mysql-bin
binlog-format = ROW
binlog-row-image = FULL

# 重启MySQL服务
systemctl restart mysqld

# 创建Debezium用户
CREATE USER ‘debezium’@’%’ IDENTIFIED BY ‘password’;
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO ‘debezium’@’%’;

# 步骤6：配置Kafka Connect
# 创建Kafka Connect配置文件
cat << EOF > /opt/kafka/config/connect-mysql-source.properties
name=mysql-source-connector
connector.class=io.debezium.connector.mysql.MySqlConnector
tasks.max=1
database.hostname=localhost
database.port=3306
database.user=debezium
database.password=password
database.server.id=184054
database.server.name=dbserver1
database.include.list=app_db
table.include.list=app_db.users
snapshot.mode=initial
EOF

# 启动Kafka Connect
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties /opt/kafka/config/connect-mysql-source.properties

# 步骤7：消费Kafka消息
# 启动Kafka消费者
$KAFKA_HOME/bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic dbserver1.app_db.users –from-beginning

# 步骤8：测试数据同步
# 在MySQL中插入数据
mysql -u root -p -e “INSERT INTO app_db.users (id, name, department) VALUES (4, ‘user4’, ‘IT’);”

# 查看Kafka消费者输出，应该能看到新插入的数据

# 步骤9：验证集成
# 查看Kafka消息
$KAFKA_HOME/bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic dbserver1.app_db.users –from-beginning

# MySQL与Kafka集成最佳实践
1. 合理配置Kafka：根据数据量和处理需求配置合适的参数
2. 启用binlog：确保MySQL启用了binlog，并且格式为ROW
3. 配置Debezium：根据实际情况配置Debezium connector
4. 监控与日志：开启监控和日志，及时发现和处理问题
5. 错误处理：配置错误处理策略，确保数据同步的可靠性
6. 测试验证：在生产环境部署前进行充分的测试
7. 性能优化：根据实际情况优化Kafka和MySQL的配置
8. 安全措施：实施安全措施，保护数据安全

# 配置Elasticsearch
export ES_HOME=/opt/elasticsearch
export PATH=$PATH:$ES_HOME/bin

# 启动Elasticsearch
$ES_HOME/bin/elasticsearch -d

# 步骤2：安装Logstash
# 下载并安装Logstash
tar -xzf logstash-7.16.2-linux-x86_64.tar.gz
mv logstash-7.16.2 /opt/logstash

# 配置Logstash
export LOGSTASH_HOME=/opt/logstash
export PATH=$PATH:$LOGSTASH_HOME/bin

# 步骤3：配置Logstash管道
# 创建Logstash配置文件
cat << EOF > /opt/logstash/config/mysql-to-es.conf
input {
jdbc {
jdbc_driver_library => “/path/to/mysql-connector-java-8.0.29.jar”
jdbc_driver_class => “com.mysql.cj.jdbc.Driver”
jdbc_connection_string => “jdbc:mysql://localhost:3306/app_db”
jdbc_user => “root”
jdbc_password => “rootpassword”
schedule => “* * * * *”
statement => “SELECT * FROM users WHERE id > :sql_last_value”
tracking_column => “id”
use_column_value => true
last_run_metadata_path => “/opt/logstash/config/mysql_last_run”
}
}

output {
elasticsearch {
hosts => [“localhost:9200”]
index => “users”
document_id => “%{id}”
}
stdout {
codec => rubydebug
}
}
EOF

# 步骤4：启动Logstash
# 启动Logstash
$LOGSTASH_HOME/bin/logstash -f /opt/logstash/config/mysql-to-es.conf

# 步骤5：使用Elasticsearch查询数据
# 查看索引
curl -X GET “localhost:9200/_cat/indices?v”

# 查询数据
curl -X GET “localhost:9200/users/_search?pretty”

# 步骤6：测试数据同步
# 在MySQL中插入数据
mysql -u root -p -e “INSERT INTO app_db.users (id, name, department) VALUES (5, ‘user5’, ‘HR’);”

# 等待Logstash同步数据
# 查询Elasticsearch数据
curl -X GET “localhost:9200/users/_search?pretty”

# 步骤7：验证集成
# 查看Elasticsearch中的数据
curl -X GET “localhost:9200/users/_search?pretty”

# MySQL与Elasticsearch集成最佳实践
1. 合理配置Logstash：根据数据量和处理需求配置合适的参数
2. 增量同步：使用增量同步，减少数据传输量
3. 索引设计：设计合理的Elasticsearch索引结构
4. 映射配置：配置合适的字段映射，提高搜索性能
5. 监控与日志：开启监控和日志，及时发现和处理问题
6. 错误处理：配置错误处理策略，确保数据同步的可靠性
7. 测试验证：在生产环境部署前进行充分的测试
8. 性能优化：根据实际情况优化Elasticsearch和MySQL的配置

# 问题描述
# 需要将MySQL中的用户数据导入Hadoop，进行离线分析

# 解决方案
## 步骤1：安装Sqoop
# 下载Sqoop
tar -xzf sqoop-1.4.7.bin__hadoop-2.6.0.tar.gz
mv sqoop-1.4.7.bin__hadoop-2.6.0 /opt/sqoop

# 配置Sqoop
export SQOOP_HOME=/opt/sqoop
export PATH=$PATH:$SQOOP_HOME/bin

# 步骤2：配置MySQL驱动
# 将MySQL驱动复制到Sqoop的lib目录
cp mysql-connector-java-8.0.29.jar /opt/sqoop/lib/

## 步骤3：从MySQL导入数据到Hive
# 使用Sqoop从MySQL导入数据到Hive
sqoop import \
–connect jdbc:mysql://localhost:3306/app_db \
–username root \
–password rootpassword \
–table users \
–hive-import \
–hive-table app_db.users \
–m 1

## 步骤4：在Hive中分析数据
# 启动Hive
hive

# 执行分析查询
SELECT department, COUNT(*) as count FROM app_db.users GROUP BY department;

# 预期输出：
+————+——-+
| department | count |
+————+——-+
| IT | 2 |
| HR | 2 |
| Finance | 1 |
+————+——-+

## 步骤5：导出分析结果到MySQL
# 在Hive中创建结果表
CREATE TABLE app_db.department_counts AS SELECT department, COUNT(*) as count FROM app_db.users GROUP BY department;

# 使用Sqoop导出结果到MySQL
sqoop export \
–connect jdbc:mysql://localhost:3306/app_db \
–username root \
–password rootpassword \
–table department_counts \
–export-dir /user/hive/warehouse/app_db.db/department_counts \
–input-fields-terminated-by ‘\001’

## 步骤6：验证集成
# 查看MySQL中的结果
mysql -u root -p -e “SELECT * FROM app_db.department_counts;”

# 预期输出：
+————+——-+
| department | count |
+————+——-+
| IT | 2 |
| HR | 2 |
| Finance | 1 |
+————+——-+

# 处理效果
# 成功将MySQL数据导入Hadoop
# 在Hive中进行了数据分析
# 将分析结果导出回MySQL
# 实现了MySQL与Hadoop的集成

# 问题描述
# 需要使用Spark分析MySQL中的销售数据，生成销售报表

# 解决方案
## 步骤1：准备数据
# 在MySQL中创建销售表
CREATE TABLE app_db.sales (id INT PRIMARY KEY, product VARCHAR(50), amount DECIMAL(10,2), sale_date DATE);

# 插入测试数据
INSERT INTO app_db.sales VALUES (1, ‘Product A’, 100.00, ‘2024-01-01’), (2, ‘Product B’, 200.00, ‘2024-01-01’), (3, ‘Product A’, 150.00, ‘2024-01-02’), (4, ‘Product C’, 300.00, ‘2024-01-02’), (5, ‘Product B’, 250.00, ‘2024-01-03’);

## 步骤2：创建Spark应用
# 创建Spark应用分析销售数据
cat << EOF > spark-sales-analysis.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import sum, month, year

spark = SparkSession.builder.appName(“Sales Analysis”).getOrCreate()

# 读取MySQL销售数据
df = spark.read.format(“jdbc”).option(“url”, “jdbc:mysql://localhost:3306/app_db”).option(“driver”, “com.mysql.cj.jdbc.Driver”).option(“dbtable”, “sales”).option(“user”, “root”).option(“password”, “rootpassword”).load()

# 分析每月销售总额
df_monthly = df.groupBy(year(“sale_date”).alias(“year”), month(“sale_date”).alias(“month”)).agg(sum(“amount”).alias(“total_sales”))
df_monthly.show()

# 分析每个产品的销售总额
df_product = df.groupBy(“product”).agg(sum(“amount”).alias(“total_sales”))
df_product.show()

# 写入分析结果到MySQL
df_monthly.write.format(“jdbc”).option(“url”, “jdbc:mysql://localhost:3306/app_db”).option(“driver”, “com.mysql.cj.jdbc.Driver”).option(“dbtable”, “monthly_sales”).option(“user”, “root”).option(“password”, “rootpassword”).mode(“overwrite”).save()

df_product.write.format(“jdbc”).option(“url”, “jdbc:mysql://localhost:3306/app_db”).option(“driver”, “com.mysql.cj.jdbc.Driver”).option(“dbtable”, “product_sales”).option(“user”, “root”).option(“password”, “rootpassword”).mode(“overwrite”).save()
EOF

## 步骤3：运行Spark应用
# 运行Spark应用
spark-submit spark-sales-analysis.py

## 步骤4：验证集成
# 查看MySQL中的分析结果
mysql -u root -p -e “SELECT * FROM app_db.monthly_sales;”

# 预期输出：
+——+——-+————+
| year | month | total_sales |
+——+——-+————+
| 2024 | 1 | 1000.00 |
+——+——-+————+

# 查看产品销售结果
mysql -u root -p -e “SELECT * FROM app_db.product_sales;”

# 预期输出：
+———-+————+
| product | total_sales |
+———-+————+
| Product A | 250.00 |
| Product B | 450.00 |
| Product C | 300.00 |
+———-+————+

# 处理效果
# 成功使用Spark分析MySQL中的销售数据
# 生成了每月销售总额和每个产品的销售总额
# 将分析结果写回MySQL
# 实现了MySQL与Spark的集成

# 问题描述
# 需要实时同步MySQL中的用户数据到Kafka，供下游系统处理

# 解决方案
## 步骤1：配置MySQL
# 启用binlog
# 编辑my.cnf文件
[mysqld]
server-id = 1
log-bin = mysql-bin
binlog-format = ROW
binlog-row-image = FULL

# 重启MySQL服务
systemctl restart mysqld

# 创建Debezium用户
CREATE USER ‘debezium’@’%’ IDENTIFIED BY ‘password’;
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO ‘debezium’@’%’;

## 步骤2：启动Kafka服务
# 启动Zookeeper
$KAFKA_HOME/bin/zookeeper-server-start.sh -daemon $KAFKA_HOME/config/zookeeper.properties

# 启动Kafka broker
$KAFKA_HOME/bin/kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties

## 步骤3：配置Debezium
# 创建Kafka Connect配置文件
cat << EOF > /opt/kafka/config/connect-mysql-source.properties
name=mysql-source-connector
connector.class=io.debezium.connector.mysql.MySqlConnector
tasks.max=1
database.hostname=localhost
database.port=3306
database.user=debezium
database.password=password
database.server.id=184054
database.server.name=dbserver1
database.include.list=app_db
table.include.list=app_db.users
snapshot.mode=initial
EOF

# 启动Kafka Connect
$KAFKA_HOME/bin/connect-standalone.sh $KAFKA_HOME/config/connect-standalone.properties /opt/kafka/config/connect-mysql-source.properties

## 步骤4：消费Kafka消息
# 启动Kafka消费者
$KAFKA_HOME/bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic dbserver1.app_db.users –from-beginning

## 步骤5：测试数据同步
# 在MySQL中插入数据
mysql -u root -p -e “INSERT INTO app_db.users (id, name, department) VALUES (6, ‘user6’, ‘IT’);”

# 在MySQL中更新数据
mysql -u root -p -e “UPDATE app_db.users SET department=’HR’ WHERE id=6;”

# 在MySQL中删除数据
mysql -u root -p -e “DELETE FROM app_db.users WHERE id=6;”

# 查看Kafka消费者输出，应该能看到插入、更新和删除的消息

## 步骤6：验证集成
# 查看Kafka消息
$KAFKA_HOME/bin/kafka-console-consumer.sh –bootstrap-server localhost:9092 –topic dbserver1.app_db.users –from-beginning

# 处理效果
# 成功将MySQL数据实时同步到Kafka
# 捕获了数据的插入、更新和删除操作
# 实现了MySQL与Kafka的集成
# 下游系统可以通过消费Kafka消息获取实时数据

# 问题描述
# 需要将MySQL中的产品数据同步到Elasticsearch，实现高效的产品搜索

# 解决方案
## 步骤1：准备数据
# 在MySQL中创建产品表
CREATE TABLE app_db.products (id INT PRIMARY KEY, name VARCHAR(100), description TEXT, price DECIMAL(10,2), category VARCHAR(50));

# 插入测试数据
INSERT INTO app_db.products VALUES (1, ‘Product A’, ‘High quality product A’, 100.00, ‘Electronics’), (2, ‘Product B’, ‘High quality product B’, 200.00, ‘Electronics’), (3, ‘Product C’, ‘High quality product C’, 150.00, ‘Clothing’), (4, ‘Product D’, ‘High quality product D’, 300.00, ‘Clothing’), (5, ‘Product E’, ‘High quality product E’, 250.00, ‘Home’);

## 步骤2：配置Logstash
# 创建Logstash配置文件
cat << EOF > /opt/logstash/config/mysql-to-es.conf
input {
jdbc {
jdbc_driver_library => “/path/to/mysql-connector-java-8.0.29.jar”
jdbc_driver_class => “com.mysql.cj.jdbc.Driver”
jdbc_connection_string => “jdbc:mysql://localhost:3306/app_db”
jdbc_user => “root”
jdbc_password => “rootpassword”
schedule => “* * * * *”
statement => “SELECT * FROM products WHERE id > :sql_last_value”
tracking_column => “id”
use_column_value => true
last_run_metadata_path => “/opt/logstash/config/mysql_last_run”
}
}

output {
elasticsearch {
hosts => [“localhost:9200”]
index => “products”
document_id => “%{id}”
}
stdout {
codec => rubydebug
}
}
EOF

## 步骤3：启动Logstash
# 启动Logstash
$LOGSTASH_HOME/bin/logstash -f /opt/logstash/config/mysql-to-es.conf

## 步骤4：使用Elasticsearch搜索数据
# 查看索引
curl -X GET “localhost:9200/_cat/indices?v”

# 搜索产品
curl -X GET “localhost:9200/products/_search?pretty” -H “Content-Type: application/json” -d ‘{
“query”: {
“match”: {
“description”: “high quality”
}
}
}’

# 按类别搜索
curl -X GET “localhost:9200/products/_search?pretty” -H “Content-Type: application/json” -d ‘{
“query”: {
“match”: {
“category”: “Electronics”
}
}
}’

## 步骤5：测试数据同步
# 在MySQL中插入新产品
mysql -u root -p -e “INSERT INTO app_db.products (id, name, description, price, category) VALUES (6, ‘Product F’, ‘High quality product F’, 350.00, ‘Electronics’);”

# 等待Logstash同步数据
# 搜索新产品
curl -X GET “localhost:9200/products/_search?pretty” -H “Content-Type: application/json” -d ‘{
“query”: {
“match”: {
“name”: “Product F”
}
}
}’

## 步骤6：验证集成
# 查看Elasticsearch中的数据
curl -X GET “localhost:9200/products/_search?pretty”

# 处理效果
# 成功将MySQL产品数据同步到Elasticsearch
# 实现了高效的产品搜索
# 支持按描述和类别搜索
# 实现了MySQL与Elasticsearch的集成