课本代码中文注释版

4-23 读取usermsg数据

// 导入Elasticsearch Spark SQL包
import org.elasticsearch.spark.sql._
// 定义Elasticsearch查询条件，查询phone_no字段值为5143217的记录

val defaultQuery: String = "?q=phone_no:5143217"
// 指定要查询的Elasticsearch表名

val esTable="usermsg"
// 配置Elasticsearch连接参数

val options =Map(

("es.nodes","hadoop5"),                                    // ES集群节点地址

("es.port","9200"),                                        // ES端口号

("es.read.metadata","false"),                              // 不读取元数据

("es.mapping.date.rich","false"),                          // 禁用富日期映射

("es.net.http.auth.user","elastic"),                       // ES认证用户名

("es.net.http.auth.pass","igeWFd7crobzLnp*=EA-"),         // ES认证密码

("es.nodes.wan.only","true")                               // 仅使用WAN节点

)
// 使用Spark从Elasticsearch读取数据并创建DataFrame

val esDf=spark.esDF(esTable, defaultQuery, options)
// 显示指定字段的查询结果

esDf.select("phone_no","owner_name","owner_code","run_name","run_time").show()

4-24 读取usermsg所有数据

// 定义查询所有数据的ES查询条件
val defaultQuery: String = "?q=*:*"
// 指定要查询的Elasticsearch表名

val esTable="usermsg"
// 配置Elasticsearch连接参数（同上）

val options =Map(

("es.nodes","hadoop5"),

("es.port","9200"),

("es.read.metadata","false"),

("es.mapping.date.rich","false"),

("es.net.http.auth.user","elastic"),

("es.net.http.auth.pass","igeWFd7crobzLnp*=EA-"),

("es.nodes.wan.only","true")

)
// 从Elasticsearch读取所有数据

val esDf=spark.esDF(esTable, defaultQuery, options)
// 生成临时表名，使用当前时间戳确保唯一性

val sparkTable: String = "tmp" + System.currentTimeMillis()
// 将DataFrame注册为临时表，便于使用SQL查询

esDf.registerTempTable(sparkTable)
// 使用SQL查询临时表，显示前10条记录

spark.sql("select * from tmp1742539402423 limit 10").show()

4-25 获取某个时间之前的数据

// 导入日期处理相关的类
import java.text.SimpleDateFormat
import java.util.{Calendar, Date, Properties}
/**

获取指定时间之前某个时间点的Date对象
@param timePattern 时间格式模式
@param timeRangeValue 时间范围数值
@param timeRangePattern 时间范围单位（Y/M/D）
@param startTime 起始时间字符串
@return 计算后的Date对象

*/

def getBeforeTime(

timePattern: String,

timeRangeValue: Int,

timeRangePattern: String,

startTime: String

): Date = {

// 创建日期格式化对象

val dateFormat = new SimpleDateFormat(timePattern)

// 创建Calendar实例用于日期计算

val cal = Calendar.getInstance()
try {

// 解析起始时间字符串并设置到Calendar中

cal.setTime(dateFormat.parse(startTime))

} catch {

// 捕获时间解析异常并抛出运行时异常

case e: Exception => throw new RuntimeException(s"时间解析失败: ${e.getMessage}")

}
// 根据时间单位进行时间计算

timeRangePattern match {

case "Y" => cal.add(Calendar.YEAR, -timeRangeValue)           // 减去指定年数

case "M" => cal.add(Calendar.MONTH, -timeRangeValue)          // 减去指定月数

case "D" => cal.add(Calendar.DAY_OF_MONTH, -timeRangeValue)   // 减去指定天数

case _ => throw new IllegalArgumentException(s"无效的时间单位: $timeRangePattern")

}
// 返回计算后的Date对象

cal.getTime

}
/**

获取指定时间之前某个时间点的字符串表示
@param timePattern 时间格式模式
@param timeRangeValue 时间范围数值
@param timeRangePattern 时间范围单位
@param startTime 起始时间字符串
@return 格式化后的时间字符串

*/

def getBeforeTimeStr(timePattern: String, timeRangeValue: Int, timeRangePattern: String, startTime: String): String = {

// 创建日期格式化对象

val dateFormat: SimpleDateFormat = new SimpleDateFormat(timePattern)

// 格式化并返回计算后的时间字符串

dateFormat.format(getBeforeTime(timePattern, timeRangeValue,timeRangePattern, startTime))

}
/**

获取当前日期的字符串表示
@param timePattern 时间格式模式
@return 当前时间的字符串表示

*/

def getNowDate(timePattern: String ): String= {

// 获取当前时间

val now: Date = new Date()

// 创建日期格式化对象

val dateFormat: SimpleDateFormat = new SimpleDateFormat(timePattern)
// 格式化当前时间

val  date = dateFormat.format(now)
// 返回格式化后的时间字符串

return date

}
// 调用函数获取3个月前的时间字符串

getBeforeTimeStr("yyyy-MM-dd HH:mm:ss",3,"M",getNowDate("yyyy-MM-dd HH:mm:ss"))

4-26 选择数据保存到Hive中

// 定义目标Hive表名
val hiveTable="zxh.usermsg111"
// 定义要选择的字段列表

val selectedCols="addressoj,estate_name,force,open_time,owner_code,owner_name,phone_no,run_name,run_time,sm_code,sm_name,terminal_no"
// 定义时间字段名，用于时间过滤

val timeColName="run_time"
// 构建CREATE TABLE AS SELECT SQL语句

// 选择指定字段并过滤50年前到现在的数据

val sql = "CREATE TABLE " +hiveTable + " as select " + selectedCols +" from " + sparkTable+ " where " + timeColName + " >' " +getBeforeTimeStr("yyyy-MM-dd HH:mm:ss",50,"Y",getNowDate("yyyy-MM-dd HH:mm:ss")) + "'"
// 执行SQL语句创建Hive表

spark.sql(sql)

4-27 获取时间范围内的表名列表

// 导入集合和日期处理相关的类
import scala.collection.mutable.ArrayBuffer
import java.text.SimpleDateFormat
import java.util.{Calendar, Date, Properties}
/**


获取两个日期之间的所有日期数组


@param start 开始日期


@param end 结束日期


@return 日期数组

*/

def getRangeDays(start: Date, end: Date): Array[Date] = {

// 创建开始日期的Calendar对象

val calBegin = Calendar.getInstance()

calBegin.setTime(start);
// 创建结束日期的Calendar对象

val calEnd = Calendar.getInstance()

calEnd.setTime(end)
// 创建可变数组存储日期

val arr = new ArrayBufferDate
// 循环添加每一天的日期，直到达到结束日期

while (end.after(calBegin.getTime())) {

arr.append(calBegin.getTime)        // 添加当前日期

calBegin.add(Calendar.DATE, 1)      // 日期加1天

}
// 转换为不可变数组并返回

arr.toArray

}


/**

获取指定时间范围内的ES表名列表
@param timePattern 时间格式模式
@param timeRangeValue 时间范围数值
@param timeRangePattern 时间范围单位
@param esIndexPre ES索引前缀
@param esIndexType ES索引时间格式
@param esType ES类型
@param startTime 起始时间
@return ES表名列表

*/

def getBeforeTimeTableNames(

timePattern: String,

timeRangeValue: Int,

timeRangePattern: String,

esIndexPre: String,

esIndexType: String,

esType: String,

startTime: String

): List[String] = {

// 计算开始时间

val start = getBeforeTime(timePattern, timeRangeValue, timeRangePattern, startTime)

// 创建日期格式化对象

val df: SimpleDateFormat = new SimpleDateFormat(timePattern)
// 解析结束时间

val end = df.parse(startTime)
// 创建ES索引日期格式化对象

val dateFormat = new SimpleDateFormat(esIndexType)
// 获取时间范围内的所有日期并格式化为ES索引格式

val days = for (d <- getRangeDays(start, end)) yield dateFormat.format(d)
// 去重并构建完整的ES表名（索引/类型格式）

days.toSet.map ((x: String) => esIndexPre + x + "/"+esType).toList

}
// 调用函数获取10天内的media_index表名列表

getBeforeTimeTableNames("yyyy-MM-dd HH:mm:ss",10,"D","media_index","yyyyww","meida","2018-08-01 00:00:00")

4-28 判断index/type是否存在于Elasticsearch数据中

// 导入Elasticsearch相关的类
import org.elasticsearch.hadoop.rest.RestClient
import org.elasticsearch.spark.cfg.SparkSettingsManager
import org.elasticsearch.spark.sql._
import scala.collection.JavaConverters._
// ES连接配置参数

val options =Map(

("es.nodes","hadoop5"),

("es.port","9200"),

("es.read.metadata","false"),

("es.mapping.date.rich","false"),

("es.net.http.auth.user","elastic"),

("es.net.http.auth.pass","igeWFd7crobzLnp*=EA-"),

("es.nodes.wan.only","true")

)
// 创建Spark设置管理器并加载配置

val settings = new SparkSettingsManager().load(spark.sparkContext.getConf).merge(options.asJava)
// 创建ES REST客户端

val client = new RestClient(settings)
// 导入HTTP请求相关的类

import java.net.{HttpURLConnection, URL}

import java.util.Base64

import scala.io.Source

import scala.util.parsing.json.JSON
// 构造获取ES索引列表的HTTP请求URL

val url = new URL("http://hadoop5:9200/_cat/indices?format=json")
// 打开HTTP连接

val conn = url.openConnection().asInstanceOf[HttpURLConnection]
// 构建Basic认证字符串

val userPass = "elastic:igeWFd7crobzLnp*=EA-"

val basicAuth = "Basic " + Base64.getEncoder.encodeToString(userPass.getBytes())
// 设置请求头和请求方法

conn.setRequestProperty("Authorization", basicAuth)

conn.setRequestMethod("GET")
// 读取响应内容

val indicesJson = Source.fromInputStream(conn.getInputStream).mkString
// 关闭连接

conn.disconnect()
// 解析JSON响应并提取索引名称

val allTables = JSON.parseFull(indicesJson) match {

case Some(list: List[Map[String, Any]] @unchecked) =>

list.map(_("index").toString)  // 提取index字段值作为表名

case _ =>

println("解析失败"); List()    // 解析失败时返回空列表

}
// 过滤出以"media_index"开头的索引，并验证是否可以正常加载

val allExistTables = allTables.filter(_.startsWith("media_index")).filter { index =>

try {

// 尝试读取索引数据作为DataFrame

val df = spark.read.format("org.elasticsearch.spark.sql").options(options).load(index)

df.columns.nonEmpty  // 检查是否有字段，有字段说明索引可用

} catch {

case e: Exception =>

// 捕获加载异常并输出错误信息

println(s"index 加载失败: {e.getMessage}")

false

}

}
// 关闭ES客户端

client.close()
// 输出所有存在且可用的表名

println("allExistTables: " + allExistTables)

4-29 获取所有index/type数据为DataFrame

// 定义要选择的字段列表（注释掉的是原始字段列表）
//val selectedCols= "audio_lang,category_name,duration,end_time,first_show_time,origin_time,owner_code,owner_name,phone_no,program_title,region,res_name,res_type,resolution,sm_name,station_name,terminal_no,vod_cat_tags,vod_title"
val selectedCols= "audio_lang,category_name,duration,end_time,first_show_time,origin_time,owner_code,owner_name,phone_no,program_title,region,res_name,res_type,resolution,sm_name,station_name,terminal_no,vod_cat_tags,vod_title"
// 将字段字符串分割为数组

val selectedColsArr = selectedCols.split(",")
// 获取第一个字段（用于select操作）

val firstCol = selectedColsArr(0).trim
// 获取其余字段（用于select操作的可变参数）

val tailCols = selectedColsArr.slice(1, selectedColsArr.length).map(_.trim)
// 遍历所有存在的表，读取数据并合并为一个DataFrame

val esDf = allExistTables.map { x =>

val index = x.split("/")(0)  // 提取索引名称（去掉type部分）

spark.esDF(index, defaultQuery, options)  // 从ES读取数据

}.reduce((x1, x2) =>

// 选择指定字段并将多个DataFrame合并

x1.select(firstCol, tailCols: _).union(x2.select(firstCol, tailCols: _))

)
// 查看DataFrame的结构信息

esDf.printSchema()
// 统计总记录数

esDf.count()
// 显示前10条记录的指定字段，不截断内容

esDf.select("terminal_no", "phone_no").show(10, false)

4-36 创建标签表user_label

-- 创建用户画像数据库，设置字符集为utf8mb4
CREATE DATABASE user_profile  CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci;
-- 切换到用户画像数据库

use user_profile;
-- 创建用户标签表

CREATE TABLE user_label (

phone_no text,        -- 用户手机号

label text,           -- 用户标签

parent_label text NOT NULL  -- 父级标签分类

) ENGINE=MyISAM DEFAULT CHARSET = utf8mb4  -- 使用MyISAM存储引擎

COLLATE = utf8mb4_unicode_ci;              -- 设置排序规则

4-37 Spark SQL查询用户标签

// 导入SparkSession
import org.apache.spark.sql.SparkSession
// 创建SparkSession实例，启用Hive支持

val spark = SparkSession.builder().appName("UserLabelExample").enableHiveSupport().getOrCreate()
// 执行SQL查询，根据用户电视入网时长进行用户分类

val data = spark.sql("""

SELECT

t1.phone_no,                    -- 用户手机号

CASE

WHEN T > 8 THEN '老用户'      -- 入网超过8年的为老用户

WHEN T > 4 AND T <= 8 THEN '中等用户'  -- 入网4-8年的为中等用户

ELSE '新用户'                  -- 入网不足4年的为新用户

END AS label,                   -- 用户标签

'电视入网程度' AS parent_label    -- 父级标签分类

FROM (

SELECT

phone_no,                     -- 用户手机号

-- 计算用户最早入网距今的年数

MAX(datediff(current_date(), open_time) / 365) AS T

FROM zxh.mediamatch_usermsg_process   -- 从处理后的用户消息表查询

WHERE sm_name LIKE '%电视%'           -- 过滤电视相关业务

AND open_time IS NOT NULL          -- 入网时间不为空

GROUP BY phone_no                     -- 按手机号分组

) t1

""")
// 显示查询结果的前5条记录

data.show(5)

4-38 将数据写入MySQL

// 导入Properties类用于数据库连接配置
import java.util.Properties  
// 设置数据写入模式为追加

val saveMode = "append"
// 设置目标表名

val outputTable = "user_label"
// 设置MySQL数据库连接URL

val url = "jdbc:mysql://hadoop1:3306/user_profile?useSSL=false&serverTimezone=UTC&useUnicode=true&characterEncoding=UTF-8"
// 创建数据库连接属性对象

val connectionProperties = new Properties()

connectionProperties.setProperty("user", "root")                    // 数据库用户名

connectionProperties.setProperty("password", "Hadoop123")           // 数据库密码

connectionProperties.setProperty("driver", "com.mysql.jdbc.Driver") // JDBC驱动类
// 将DataFrame数据写入MySQL表

data.write.mode(saveMode).jdbc(url, outputTable, connectionProperties)

4-41 数据处理和提交作业

// 读取原始用户消息表数据
val originUsermsg = spark.sql("select * from zxh.mediamatch_usermsg")
// 读取处理后的用户消息表数据

val processedUsermsg = spark.sql("select * from zxh.mediamatch_usermsg_process")
// 统计处理后表的记录数

processedUsermsg.count

# 使用spark-submit提交数据处理作业
spark-submit \
--class com.tipdm.scala.chapter_3_6_processing.DataProcess \  # 指定主类
--master yarn \                                               # 使用YARN作为集群管理器
--deploy-mode cluster \                                       # 使用集群部署模式
--executor-memory 2G \                                        # 设置executor内存为2G
--num-executors 2 \                                          # 设置executor数量为2个
data-process.jar \                                           # 应用程序JAR包
zxh.media_index_3m \                                         # 输入表1：3个月媒体索引表
zxh.media_index_3m_process \                                 # 输出表1：处理后的3个月媒体索引表
zxh.mediamatch_userevent \                                   # 输入表2：用户事件表
zxh.mediamatch_userevent_process \                           # 输出表2：处理后的用户事件表
zxh.mediamatch_usermsg \                                     # 输入表3：用户消息表
zxh.mediamatch_usermsg_process \                             # 输出表3：处理后的用户消息表
zxh.mmconsume_billevents \                                   # 输入表4：计费事件表
zxh.mmconsume_billevent_process \                            # 输出表4：处理后的计费事件表
zxh.order_index_v3 \                                         # 输入表5：订单索引表v3
zxh.order_index_process                                      # 输出表5：处理后的订单索引表

# 另一种spark-submit命令格式
spark-submit \
--class DataProcess \                                        # 指定主类
--master yarn \                                             # 使用YARN作为集群管理器
--deploy-mode cluster \                                     # 使用集群部署模式
--name DataProcess \                                        # 设置应用程序名称
/root/bigdata/4/elasticsearch-spark-20_2.11-8.17.3.jar \   # Elasticsearch Spark连接器JAR包
/root/chapter4.1-1.0-SNAPSHOT.jar \                        # 应用程序JAR包
zxh.media_index_3m \                                        # 参数1：输入表
zxh.media_index_3m_process \                                # 参数2：输出表
zxh.mediamatch_userevent \                                  # 参数3：输入表
zxh.mediamatch_userevent_process \                          # 参数4：输出表
zxh.mediamatch_usermsg \                                    # 参数5：输入表
user_profile.mediamatch_usermsg_process \                   # 参数6：输出表（用户画像库）
user_profile.mmconsume_billevents \                         # 参数7：输入表（用户画像库）
user_profile.mmconsume_billevent_process \                  # 参数8：输出表（用户画像库）
user_profile.order_index_v3 \                               # 参数9：输入表（用户画像库）
user_profile.order_index_process                            # 参数10：输出表（用户画像库）