# 导入数据到HDFS
sqoop import \
  --connect jdbc:mysql://192.168.1.13:4000/fgedudb \
  --username root \
  --password root123 \
  --table fgedu_users \
  --target-dir /user/hadoop/fgedu_users \
  --delete-target-dir \
  --num-mappers 4

# 查看导入的数据
hadoop fs -ls /user/hadoop/fgedu_users
hadoop fs -cat /user/hadoop/fgedu_users/part-m-00000 | head -10

Found 5 items
-rw-r--r--   3 hadoop supergroup          0 2024-04-09 10:00 /user/hadoop/fgedu_users/_SUCCESS
-rw-r--r--   3 hadoop supergroup       1024 2024-04-09 10:00 /user/hadoop/fgedu_users/part-m-00000
-rw-r--r--   3 hadoop supergroup        987 2024-04-09 10:00 /user/hadoop/fgedu_users/part-m-00001
-rw-r--r--   3 hadoop supergroup       1056 2024-04-09 10:00 /user/hadoop/fgedu_users/part-m-00002
-rw-r--r--   3 hadoop supergroup        978 2024-04-09 10:00 /user/hadoop/fgedu_users/part-m-00003

1,admin,admin@example.com,30,2024-01-01 00:00:00
2,user1,user1@example.com,25,2024-01-02 00:00:00
3,user2,user2@example.com,35,2024-01-03 00:00:00
4,user3,user3@example.com,28,2024-01-04 00:00:00
5,user4,user4@example.com,32,2024-01-05 00:00:00

3.4 TiDB与Hive集成

创建Hive外部表映射TiDB数据

# 在Hive中创建外部表
CREATE EXTERNAL TABLE fgedu_users (
  id INT,
  username STRING,
  email STRING,
  age INT,
  created_at TIMESTAMP
) 
ROW FORMAT DELIMITED 
FIELDS TERMINATED BY ',' 
LOCATION '/user/hadoop/fgedu_users';

# 查询Hive表
SELECT * FROM fgedu_users LIMIT 10;

# 执行分析
SELECT age, COUNT(*) FROM fgedu_users GROUP BY age;

+-----+----------+------------------+-----+---------------------+
| id  | username | email            | age | created_at          |
+-----+----------+------------------+-----+---------------------+
| 1   | admin    | admin@example.com | 30  | 2024-01-01 00:00:00 |
| 2   | user1    | user1@example.com | 25  | 2024-01-02 00:00:00 |
| 3   | user2    | user2@example.com | 35  | 2024-01-03 00:00:00 |
| 4   | user3    | user3@example.com | 28  | 2024-01-04 00:00:00 |学习交流加群风哥QQ113257174
| 5   | user4    | user4@example.com | 32  | 2024-01-05 00:00:00 |
+-----+----------+------------------+-----+---------------------+

+-----+--------+
| age | _c1    |
+-----+--------+
| 25  | 1      |
| 28  | 1      |
| 30  | 1      |
| 32  | 1      |
| 35  | 1      |
+-----+--------+

# 创建TiCDC同步任务
tiup ctl:v7.5.0 cdc changefeed create --pd=http://192.168.1.10:2379 --sink-uri="kafka://192.168.1.20:9092/orders?protocol=canal-json" --config=cdc-orders.yaml

# cdc-orders.yaml配置示例
[cdc]
sink-uri = "kafka://192.168.1.20:9092/orders?protocol=canal-json"

[filter]
rules = ["fgedudb.fgedu_orders"]

# Spark Streaming代码
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._

val spark = SparkSession.builder()
  .appName("Order Analysis")
  .getOrCreate()

// 读取Kafka数据
val df = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "192.168.1.20:9092")
  .option("subscribe", "orders")
  .load()

// 解析JSON数据
val parsedDF = df.selectExpr("CAST(value AS STRING)")
  .select(from_json(col("value"), schema).as("data"))
  .select("data.*")

// 实时分析
val orderStats = parsedDF
  .groupBy(window(col("created_at"), "1 minute"), col("status"))
  .count()

// 输出结果
orderStats.writeStream
  .outputMode("complete")
  .format("console")
  .start()
  .awaitTermination()

# 1. 使用TiCDC将TiDB数据同步到Kafka
# 2. 使用Flink将Kafka数据处理后写入HDFS
# 3. 在Hive中创建外部表映射HDFS数据
# 4. 使用Spark SQL进行数据分析
# 5. 将分析结果写回TiDB或其他存储系统

# 创建Hive外部表
CREATE EXTERNAL TABLE fgedu_orders (
  id BIGINT,
  order_no STRING,
  user_id BIGINT,
  total_amount DECIMAL(12,2),
  status TINYINT,
  created_at TIMESTAMP
) 
PARTITIONED BY (date STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS PARQUET
LOCATION '/user/hadoop/orders';

# 添加分区
ALTER TABLE fgedu_orders ADD PARTITION (date='2024-04-09') LOCATION '/user/hadoop/orders/date=2024-04-09';

# Spark SQL分析
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("Sales Analysis")
  .enableHiveSupport()
  .getOrCreate()

// 执行分析
val salesAnalysis = spark.sql("""
  SELECT 
    date,
    SUM(total_amount) AS total_sales,
    COUNT(*) AS order_count,
    AVG(total_amount) AS avg_order_amount
  FROM fgedu_orders
  GROUP BY date
  ORDER BY date
""")

// 查看结果
salesAnalysis.show()

// 写入结果到TiDB
salesAnalysis.write
  .format("tidb")
  .option("tidb.addr", "192.168.1.13")
  .option("tidb.port", "4000")
  .option("tidb.user", "root")
  .option("tidb.password", "root123")
  .option("database", "fgedudb")
  .option("table", "sales_analysis")
  .mode("overwrite")
  .save()

+----------+-----------+-----------+------------------+
|      date|total_sales|order_count|avg_order_amount  |
+----------+-----------+-----------+------------------+
|2024-04-09|   15000.00|        100|           150.00|
|2024-04-10|   18000.00|        120|           150.00|
|2024-04-11|   20000.00|        130|           153.85|
+----------+-----------+-----------+------------------+