# 1. 需求分析
– 业务需求
– 实时采集应用日志
– 存储到HDFS
– 支持数据分析
– 技术需求
– 使用Flume采集日志
– 存储到HDFS
– 按日期分区存储
# 2. 方案设计
– Flume Agent配置
– Source：Exec Source
– Channel：Memory Channel
– Sink：HDFS Sink
– HDFS目录结构
– /fgedu/logs/yyyyMMdd/
– 按日期分区
– 按小时滚动
# 3. 实施步骤
– 配置Flume Agent
$ vi $FLUME_HOME/conf/fgedu_log.conf
a1.sources = r1
a1.sinks = k1
a1.channels = c1
a1.sources.r1.type = exec
a1.sources.r1.command = tail -F /var/log/fgedu/application.log
a1.sinks.k1.type = hdfs
a1.sinks.k1.hdfs.path = hdfs://fgedu.localhost:9000/fgedu/logs/%Y%m%d/%H
a1.sinks.k1.hdfs.fileType = DataStream
a1.sinks.k1.hdfs.writeFormat = Text
a1.sinks.k1.hdfs.useLocalTimeStamp = true
a1.sinks.k1.hdfs.rollInterval = 3600
a1.sinks.k1.hdfs.rollSize = 134217728
a1.sinks.k1.hdfs.rollCount = 0
a1.channels.c1.type = memory
a1.channels.c1.capacity = 10000
a1.channels.c1.transactionCapacity = 1000
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
– 启动Flume Agent
$ flume-ng agent –conf $FLUME_HOME/conf –conf-file $FLUME_HOME/conf/fgedu_log.conf –name a1 -Dflume.root.logger=INFO,console
# 4. 验证结果
– 写入日志
$ echo “2024-01-10 10:00:00 INFO [main] Application started” >> /var/log/fgedu/application.log
– 查看HDFS
$ hdfs dfs -ls /fgedu/logs/20240110/10
# 输出结果
# Found 1 items
# -rw-r–r– 1 root supergroup 45 2024-01-10 10:00 /fgedu/logs/20240110/10/FlumeData.1234567890
– 查看日志内容
$ hdfs dfs -cat /fgedu/logs/20240110/10/FlumeData.1234567890
# 输出结果
# 2024-01-10 10:00:00 INFO [main] Application started
# 5. 实施结果
– 日志采集成功
– 数据存储正常
– 分区存储正常

# 1. 需求分析
– 业务需求
– 实时处理订单数据
– 计算订单统计指标
– 实时监控和决策
– 技术需求
– 使用Kafka接收订单数据
– 使用Spark Streaming处理数据
– 存储到DM数据库
# 2. 方案设计
– Kafka Topic
– Topic名称：fgedu_order
– 分区数：3
– 副本数：2
– Spark Streaming
– 批处理间隔：10秒
– 处理逻辑：统计订单数量和金额
– 输出：写入DM数据库
# 3. 实施步骤
– 创建Kafka Topic
$ kafka-topics.sh –create –topic fgedu_order –partitions 3 –replication-factor 2 –bootstrap-server fgedu.localhost:9092
– Spark Streaming处理代码
$ vi OrderStreaming.scala
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._
import org.apache.kafka.common.serialization.StringDeserializer
import java.sql.{DriverManager, Connection}
object OrderStreaming {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName(“OrderStreaming”)
val ssc = new StreamingContext(conf, Seconds(10))
val kafkaParams = Map[String, Object](
“bootstrap.servers” -> “fgedu.localhost:9092”,
“key.deserializer” -> classOf[StringDeserializer],
“value.deserializer” -> classOf[StringDeserializer],
“group.id” -> “fgedu_group”,
“auto.offset.reset” -> “latest”,
“enable.auto.commit” -> (false: java.lang.Boolean)
)
val topics = Array(“fgedu_order”)
val stream = KafkaUtils.createDirectStream[String, String](
ssc,
PreferConsistent,
Subscribe[String, String](topics, kafkaParams)
)
val orders = stream.map(record => record.value)
val orderStats = orders.map(order => {
val fields = order.split(“,”)
(fields(0).toInt, fields(3).toDouble)
}).reduceByKey(_ + _)
orderStats.foreachRDD { rdd =>
rdd.foreachPartition { partition =>
val conn = DriverManager.getConnection(“jdbc:dm://fgedu.localhost:5236/fgedu”, “fgedu_user”, “fgedu_password”)
partition.foreach { case (user_id, total_amount) =>
val stmt = conn.prepareStatement(“INSERT INTO fgedu_order_stats (user_id, total_amount, stat_time) VALUES (?, ?, SYSDATE)”)
stmt.setInt(1, user_id)
stmt.setDouble(2, total_amount)
stmt.executeUpdate()
}
conn.close()
}
}
ssc.start()
ssc.awaitTermination()
}
}
– 编译和运行
$ spark-submit –class OrderStreaming –master local[2] OrderStreaming.jar
# 4. 验证结果
– 发送订单数据
$ kafka-console-producer.sh –broker-list fgedu.localhost:9092 –topic fgedu_order
> 1,1,1,100.00,2024-01-10 10:00:00
> 1,2,2,200.00,2024-01-10 10:00:01
> 2,3,3,300.00,2024-01-10 10:00:02
– 查询统计结果
SQL> SELECT * FROM fgedu_order_stats;
# 输出结果
# USER_ID TOTAL_AMOUNT STAT_TIME
# ——– ————- ——————-
# 1 300.00 2024-01-10 10:00:10
# 2 300.00 2024-01-10 10:00:10
# 5. 实施结果
– 实时处理成功
– 统计计算正常
– 数据存储正常

# 1. 需求分析
– 业务需求
– 整合多个数据源
– 支持数据分析
– 支持机器学习
– 技术需求
– 使用HDFS存储数据
– 使用Hive管理数据
– 使用Spark处理数据
# 2. 方案设计
– 数据湖架构
– 原始数据层（Raw）
– 清洗数据层（Cleaned）
– 聚合数据层（Aggregated）
– 数据处理流程
– 数据采集：Flume + Sqoop
– 数据处理：Spark
– 数据存储：HDFS + Hive
# 3. 实施步骤
– 创建HDFS目录
$ hdfs dfs -mkdir -p /fgedu/lake/raw
$ hdfs dfs -mkdir -p /fgedu/lake/cleaned
$ hdfs dfs -mkdir -p /fgedu/lake/aggregated
– 创建Hive数据库
hive> CREATE DATABASE fgedu_lake;
– 创建原始数据表
hive> CREATE TABLE fgedu_lake.raw_user (
user_id INT,
user_name STRING,
user_email STRING,
user_status INT,
create_time STRING,
update_time STRING
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ‘,’
STORED AS TEXTFILE
LOCATION ‘/fgedu/lake/raw/user’;
– 创建清洗数据表
hive> CREATE TABLE fgedu_lake.cleaned_user (
user_id INT,
user_name STRING,
user_email STRING,
user_status INT,
create_date DATE,
update_date DATE
)
PARTITIONED BY (load_date STRING)
STORED AS PARQUET
LOCATION ‘/fgedu/lake/cleaned/user’;
– 创建聚合数据表
hive> CREATE TABLE fgedu_lake.aggregated_user_stats (
user_id INT,
order_count INT,
total_amount DECIMAL(10,2),
avg_amount DECIMAL(10,2),
stat_date DATE
)
PARTITIONED BY (load_date STRING)
STORED AS PARQUET
LOCATION ‘/fgedu/lake/aggregated/user_stats’;
# 4. 数据处理
– 数据清洗
$ spark-submit –class DataCleaning –master yarn DataCleaning.jar
– 数据聚合
$ spark-submit –class DataAggregation –master yarn DataAggregation.jar
# 5. 验证结果
– 查询清洗数据
hive> SELECT * FROM fgedu_lake.cleaned_user WHERE load_date = ‘20240110’ LIMIT 10;
# 输出结果
# OK
# 1 fgedu_user1 fgedu_user1@fgedu.net.cn 1 2024-01-01 2024-01-10
# 2 fgedu_user2 fgedu_user2@fgedu.net.cn 1 2024-01-02 2024-01-10
# 3 user3 user3@fgedu.net.cn 1 2024-01-03 2024-01-10
– 查询聚合数据
hive> SELECT * FROM fgedu_lake.aggregated_user_stats WHERE load_date = ‘20240110’ LIMIT 10;
# 输出结果
# OK
# 1 10 1000.00 100.00 2024-01-10
# 2 5 500.00 100.00 2024-01-10
# 3 3 300.00 100.00 2024-01-10
# 6. 实施结果
– 数据湖建设成功
– 数据整合完成
– 数据处理正常