大数据开发期末代码

课本代码中文注释版

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","<已移除敏感信息>"),         // 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","<已移除敏感信息>"),
  ("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 ArrayBuffer[Date]()
     
     // 循环添加每一天的日期，直到达到结束日期
     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","<已移除敏感信息>"),
  ("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", "<已移除敏感信息>")           // 数据库密码
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：输出表（用户画像库）