# 检查集群硬件配置
[root@fgedu.net.cn ~]# hadoop dfsadmin -report
Configured Capacity: 1099511627776 (1024.0 GB)
Present Capacity: 879609302016 (819.2 GB)
DFS Remaining: 703687441664 (655.4 GB)
DFS Used: 175921860352 (163.8 GB)
DFS Used%: 20.0%
Under replicated blocks: 0
Blocks with corrupt replicas: 0
Missing blocks: 0

# 创建HDFS目录结构
[root@fgedu.net.cn ~]# hdfs dfs -mkdir -p /user/fgedu/signaling/raw
[root@fgedu.net.cn ~]# hdfs dfs -mkdir -p /user/fgedu/signaling/processed
[root@fgedu.net.cn ~]# hdfs dfs -mkdir -p /user/fgedu/signaling/analysis
[root@fgedu.net.cn ~]# hdfs dfs -mkdir -p /user/fgedu/signaling/archive

# 设置目录权限
[root@fgedu.net.cn ~]# hdfs dfs -chmod -R 755 /user/fgedu/signaling

# 使用Flume采集信令数据
[root@fgedu.net.cn ~]# vi /bigdata/app/flume/conf/signaling.conf

agent.sources = tail
agent.sinks = kafka
agent.channels = memory

agent.sources.tail.type = exec
agent.sources.tail.command = tail -F /var/log/signaling/signaling.log
agent.sources.tail.channels = memory

agent.channels.memory.type = memory
agent.channels.memory.capacity = 10000
agent.channels.memory.transactionCapacity = 1000

agent.sinks.kafka.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.kafka.topic = signaling
agent.sinks.kafka.brokerList = fgedu.net.cn:9092
agent.sinks.kafka.batchSize = 1000
agent.sinks.kafka.channel = memory

# 使用Spark进行数据清洗
[root@fgedu.net.cn ~]# spark-shell

scala> import org.apache.spark.sql.SparkSession
scala> import org.apache.spark.sql.functions._

scala> val spark = SparkSession.builder().appName(“SignalingCleaning”).getOrCreate()

scala> val signaling = spark.read.text(“/user/fgedu/signaling/raw”)

scala> // 解析信令数据
scala> val parsedSignaling = signaling
.withColumn(“message”, split(col(“value”), “\\|”))
.withColumn(“timestamp”, element_at(col(“message”), 1))
.withColumn(“imsi”, element_at(col(“message”), 2))
.withColumn(“imei”, element_at(col(“message”), 3))
.withColumn(“msisdn”, element_at(col(“message”), 4))
.withColumn(“event_type”, element_at(col(“message”), 5))
.withColumn(“cell_id”, element_at(col(“message”), 6))
.withColumn(“duration”, element_at(col(“message”), 7))
.drop(“value”, “message”)

// 数据清洗
val cleanedSignaling = parsedSignaling
.filter(col(“imsi”).isNotNull && col(“msisdn”).isNotNull)
.filter(col(“duration”).cast(“int”) > 0)

// 保存清洗后的数据
cleanedSignaling.write.parquet(“/user/fgedu/signaling/processed”)

# 使用Hive创建信令数据仓库
hive> CREATE EXTERNAL TABLE signaling (
> timestamp STRING,
> imsi STRING,
> imei STRING,
> msisdn STRING,
> event_type STRING,
> cell_id STRING,
> duration INT
> )
> STORED AS PARQUET
> LOCATION ‘/user/fgedu/signaling/processed’;

# 分析信令数据
hive> SELECT
> event_type,
> COUNT(*) AS event_count,
> AVG(duration) AS avg_duration
> FROM signaling
> GROUP BY event_type
> ORDER BY event_count DESC;