day09_实时类标签/指标

一、日志数据实时采集

2、Flume简介

2.3 项目日志数据采集Flume配置

zookeeper、Kafka的启动命令

启动zookeeper（没有启动的，才需要执行）
/export/server/zookeeper/bin/zkServer.sh start

启动Kafka
cd /export/server/kafka/bin
nohup ./kafka-server-start.sh ../config/server.sql 2>&1 &

Kafka其他的相关命令
cd /export/server/kafka/bin
查看当前集群有哪些Topic
./kafka-topics.sh --list --bootstrap-server up01:9092
新建Topic（分区数没要求，副本数<=broker节点个数）
./kafka-topics.sh --create --bootstrap-server up01:9092 --topic xtzg_nginx_log
参看Topic的详细信息
./kafka-topics.sh --describe --bootstrap-server up01:9092 --topic xtzg_nginx_log

注意: 要提前创建好Kafka的Topic

2.3.1 涉及的Flume组件和参数

• source

type: 类型，固定值TAILDIR。能同时监控一个目录或者多个文件，也能动态监控每个文件的变化，还支持断点续传，不会出现重复消费问题。
fiilegroups:  以空格分隔的文件组列表。每个文件组表示一组要跟踪的文件。
filegroups.<filegroupName>: 文件组的绝对路径。正则表达式(而不是文件系统模式)只能用于文件名。
positionFile: JSON格式的文件，记录每个文件的inode、绝对路径和最后位置。

注意: type的TAILDIR大小写不能随便写

• channel

type: 类型，固定值 org.apache.flume.channel.kafka.KafkaChannel
kafka.bootstrap.servers: Kafka集群中的broker列表。格式：hostname:port，多个用逗号隔开。
kafka.topic: channel要用的topic
parseAsFlumeEvent: 是否需要对采集到的数据解析为Event对象，然后在内容前面增加topic前缀，会导致后续的内容会有部分缺失的情况。一般是false

补充：

如果采集到的数据最终想要输出到Kafka中，可以直接选择使用Kafka Channel。
注意: Kafka Channel和Kafka Sink，虽然都是将数据输出到Kafka中，但是两者的配置参数有区别

2.3.2 Nginx日志采集

• 创建nginx_to_kafka.conf文件

• nginx_to_kafka.conf配置文件内容如下

#定义组件
a1.sources = r1
a1.channels = c1

#配置source
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /export/data/workspace/user_profile/log_generate/datacollection/source_data/access-nginx.*
a1.sources.r1.positionFile = /export/data/flume/nginx_position.json


#配置channel
a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = up01:9092
a1.channels.c1.kafka.topic = xtzg_nginx_log
a1.channels.c1.parseAsFlumeEvent = false

#组装 
a1.sources.r1.channels = c1

注意: 
	1- a1.sources.r1.filegroups.f1该参数值要改成你自己的路径
	2- 文件的模糊匹配的正则表达式中写的是.*表示匹配任意内容

将上面的配置文件复制到/export/server/flume/conf
cp /export/data/workspace/user_profile/scripts/flume/nginx_to_kafka.conf /export/server/flume/conf

• 在Kafka上创建topic（前提开启zk,kafka）

cd /export/server/kafka/bin

./kafka-topics.sh --create --bootstrap-server up01:9092 --topic xtzg_nginx_log

• 启动Flume

cd /export/server/flume

bin/flume-ng agent -n a1 -c conf/ -f conf/nginx_to_kafka.conf

• 查看Kafka中的数据

cd /export/server/kafka/bin

./kafka-console-consumer.sh --bootstrap-server up01:9092 --topic xtzg_nginx_log

• 启动

运行python中的NginxLogSimulationData.py。查看kafka中数据变化，如果看到新增数据则配置成功。确认无误后关停Flume采集任务。

2.3.3 用户行为日志采集

• 创建user_event_to_kafka.conf文件
  

• user_event_to_kafka.conf配置文件内容如下

#定义组件
a1.sources = r1
a1.channels = c1

#配置source
a1.sources.r1.type = TAILDIR
a1.sources.r1.filegroups = f1
a1.sources.r1.filegroups.f1 = /export/data/workspace/user_profile/log_generate/datacollection/source_data/user-event.*
a1.sources.r1.positionFile = /export/data/flume/user_event_position.json


#配置channel
a1.channels.c1.type = org.apache.flume.channel.kafka.KafkaChannel
a1.channels.c1.kafka.bootstrap.servers = up01:9092
a1.channels.c1.kafka.topic = xtzg_user_event
a1.channels.c1.parseAsFlumeEvent = false

#组装 
a1.sources.r1.channels = c1

• 在Kafka上创建topic（前提开启zk,kafka）

cd /export/server/kafka

bin/kafka-topics.sh --create --bootstrap-server up01:9092 --topic xtzg_user_event --partitions 1 --replication-factor 1

• 启动Flume

cd /export/server/flume

bin/flume-ng agent -n a1 -c conf/ -f conf/user_event_to_kafka.conf

• 查看Kafka中的数据

cd /export/server/kafka

bin/kafka-console-consumer.sh --bootstrap-server up01:9092 --from-beginning --topic xtzg_user_event

• 启动

运行python中的EventSimulationJsonData.py。查看kafka中数据变化，如果看到新增数据则配置成功。确认无误后关停Flume采集任务。

二、Nginx日志数据统计

1、日志格式说明

​ Nginx（发音 恩几可使）是异步框架的网页服务器，也可以用作反向代理、负载平衡器和HTTP缓存。该软件由俄罗斯程序员伊戈尔·赛索耶夫（Игорь Сысоев）开发并于2004年首次公开发布

• Nginx日志包含access_log和error_log两种类型日志数据。项目中分析的数据为：access_log
• Nginx开源官网：https://nginx.org/
• 项目采集Nginx数据格式。以下为一条Nginx日志:

116.85.48.25 - - [12/Nov/2024:11:36:46 +0800] "GET /login.html HTTP/1.1" 404 729 "https://xtx.itcast.cn/referAFriend.html" "Mozilla/5.0 (iPhone; CPU iPhone OS 14_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Mobile/15E148 MicroMessenger/7.0.18(0x17001233) NetType/WIFI Language/zh_CN" "-"


Nginx日志格式说明:
	116.85.48.25: 用户访问IP地址
	- - : 用户标识(cookie信息)
	[14/Jul/2022:17:40:41 +0800]:  访问时间 + 时区
	GET : 请求方式
	/css/40.30d6d2b.css: 请求资源
	HTTP/1.1 : 请求的协议
	500 : 请求的状态码 (500 服务器错误,  200 成功  302 重定向  404 访问到未知资源)
	951 : 响应返回的字节大小
	"https://www.htv.com/official/component?WT.mc_id=3" : 来源的URL(从那个地方跳转到此页面)
	"Mozilla/5......:  浏览器标识

2、数据ETL

2.1 日志抽取

2.1.1 正则表达式

Java版本:
(?<ip>\d+\.\d+\.\d+\.\d+) (- - \[)(?<datetime>[\s\S]+)(?<t1>\][\s"]+)(?<request>[A-Z]+) (?<url>[\S]*) (?<protocol>[\S]+)["] (?<code>\d+) (?<sendbytes>\d+) ["](?<refferer>[\S]*) ["](?<useragent>[\S\s]+)["] ["](?<proxyaddr>[\S\s]+)["]

Python版本:
(?P<ip>.*?) - - \[(?P<time>.*?)\] "(?P<request>.*?)" (?P<status>.*?) (?P<bytes>.*?) "(?P<referer>.*?)" "(?P<ua>.*?)" "(?P<proxy_address>.*)"

2.1.2 基于Spark实现Nginx数据匹配

代码实现:

from pyspark.sql import SparkSession
import os
import pyspark.sql.functions as F
from pyspark.sql.types import StringType

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

if __name__ == '__main__':
    # 1- 创建SparkSession对象
    spark = SparkSession.builder\
        .appName("nginx_etl")\
        .master("local[*]")\
        .config("spark.sql.shuffle.partitions",2)\
        .getOrCreate()

    # 2- 数据输入：读取Kafka中的数据
    """
        "startingOffsets","earliest"：该配置，在实际工作中一般不需要配置。这里是为了开发代码方便
    """
    init_df = spark.readStream.format("kafka")\
        .option("kafka.bootstrap.servers","192.168.88.166:9092")\
        .option("subscribe","xtzg_nginx_log")\
        .option("startingOffsets","earliest")\
        .load()

    # 结构化流中不能以show()方式打印数据数据内容
    # init_df.show()

    # 3- 数据ETL处理
    # 3.1- value字段解码的操作
    """
        cast(StringType())：将字段数据类型强制转换为字符串。等同于SQL语句中的cast(value as string)
        
        下面两种方式都可以，推荐使用第一种，因为性能更好
    """
    # type_cast_df = init_df.select(init_df.value.cast(StringType()).alias("value"))
    type_cast_df = init_df.selectExpr("cast(value as string) as value")

    # 3.2- 通过正则表达式提取Nginx的字段
    pattern = '(?<ip>\d+\.\d+\.\d+\.\d+) (- - \[)(?<datetime>[\s\S]+)(?<t1>\][\s"]+)(?<request>[A-Z]+) (?<url>[\S]*) (?<protocol>[\S]+)["] (?<code>\d+) (?<sendbytes>\d+) ["](?<refferer>[\S]*) ["](?<useragent>[\S\s]+)["] ["](?<proxyaddr>[\S\s]+)["]'
    regexp_df = type_cast_df.select(
        F.regexp_extract("value",pattern,1).alias("ip"),
        F.regexp_extract("value",pattern,3).alias("datetime"),
        F.regexp_extract("value",pattern,4).alias("t1"),
        F.regexp_extract("value",pattern,5).alias("request"),
        F.regexp_extract("value",pattern,6).alias("url"),
        F.regexp_extract("value",pattern,7).alias("protocol"),
        F.regexp_extract("value",pattern,8).alias("code"),
        F.regexp_extract("value",pattern,9).alias("sendbytes"),
        F.regexp_extract("value",pattern,10).alias("refferer"),
        F.regexp_extract("value",pattern,11).alias("useragent"),
        F.regexp_extract("value",pattern,12).alias("proxyaddr")
    )

    # 4- 数据输出，启动流式任务
    regexp_df.writeStream.format("console").outputMode("append").start().awaitTermination()

运行结果截图：

可能遇到的错误：

原因: regexp_extract函数只能传递Java版的正则表达式，不能用Python的

2.2 字段解析

需求：根据nginx日志，ip标识唯一的用户，需要ip分组，统计得到用户访问的pv、uv、区域、状态码、终端设备的操作系统、设备品牌、浏览器、访问时间(年-月-日 时:分:秒)

2.2.1 日期格式转换

Python的datetime函数库

• 相关函数:
  • strftime(): 把日期对象转成指定的时间格式的字符串
  • strptime(): 把指定格式的日期字符串转换为日期对象
• 参考文档: https://docs.python.org/zh-cn/3/library/datetime.html#strftime-strptime-behavior
• 解析格式: %d/%b/%Y:%H:%M:%S %z => %Y-%m-%d %H:%M:%S
  • 28/Jul/2022:16:22:07 +0800 => 日期对象 => 2022-07-28 16:22:07

• 测试代码

Python方式

from datetime import datetime

if __name__ == '__main__':
    date_str = "11/Feb/2025:14:34:49 +0800"

    print(datetime.strptime(date_str, "%d/%b/%Y:%H:%M:%S %z").strftime("%Y-%m-%d %H:%M:%S"))

SparkSQL方式（重点掌握）

regexp_df.withColumn("datetime",F.from_unixtime(F.unix_timestamp("datetime","dd/MMM/yyyy:HH:mm:ss Z"),"yyyy-MM-dd HH:mm:ss"))

2.2.2 IP解析地理位置（了解）

根据IP解析地理位置

• 方式一: 使用ip解析地理位置API
  • ip地址：http://opendata.baidu.com/api.php?query=117.136.12.79&co=&resource_id=6006&oe=utf8
  • 像百度地图开发平台 / 高德地图开放平台 … 都会提供IP解析的服务接口
  • 百度地图：https://lbs.baidu.com/faq/api?title=webapi/ip-api-base
  • 高德地图：https://lbs.amap.com/api/webservice/guide/api/ipconfig
  • 其他平台：https://www.nowapi.com/
• 方式二: （了解）使用geo_ip依赖包和GeoLite2-City.mmdb库
  • 依赖包：geoip2~=4.5.0
  • 下载地址：https://gitcode.com/crownp/geolite2_demo/blob/master/src/main/resources/GeoLite2-City.mmdb

• IP在线解析测试代码

Python的Requests库的介绍：https://requests.readthedocs.io/en/latest/

#!/usr/bin/env python
# @desc : 
__coding__ = "utf-8"
__author__ = "bytedance"

import requests

def parse_ip(ip_str):
    params = {
        "query": ip_str,
        "co": "",
        "resource_id": "6006",
        "oe": "utf8",
    }
    # 发送请求
    response = requests.get(url="https://opendata.baidu.com/api.php", params=params)
    # 解析响应内容
    result = response.json()

    status = result['status']
    if status == '0':
        # 正常
        try:
            return result['data'][0]['location'].split(" ")[0]
        except:
            return "未知区域"
    else:
        return "未知区域"


if __name__ == '__main__':
    ip_str = "127.0.0.1"
    ip_str = "10.254.1.97"
    ip_str = "157.148.69.76"

    area = parse_ip(ip_str)

    print(area)

2.2.3 UA解析

UA说明

• UA为useragent简称，特指用户访问系统使用的客户端信息，一般包含操作系统，浏览器，设备品牌信息等
• UA字符串信息：http://useragentstring.com/
• 使用，需导入UA解析依赖包：from user_agents import parse
• UA的作用
  • 1.客户端识别：通过User-Agent，服务器能够识别客户端的类型和版本，从而提供相应的内容和服务。比如，在移动设备上展示适合屏幕大小的网页布局，或在不同浏览器上提供兼容性优化。
  • 2.统计分析：网站和应用开发者可以利用User-Agent来收集客户端的信息，进行用户行为分析和统计。这有助于了解用户使用的设备和偏好，以便进行产品和服务的改进。
  • 3.安全性：User-Agent也可以用于安全验证和防止恶意行为。通过分析User-Agent，服务器可以检测到异常或伪造的请求，并采取相应的安全措施。

• 测试代码:

from user_agents import parse

if __name__ == '__main__':
    ua_str = "Mozilla/5.0 (iPhone; CPU iPhone OS 13_6_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/13.0 MQQBrowser/11.0.7 Mobile/15E148 Safari/604.1 QBWebViewUA/2 QBWebViewType/1 WKType/1"

    result = parse(ua_str)

    # os操作系统信息
    print("os----------")
    print(result.os.family)
    print(result.os.version)
    print(result.os.version_string)

    # brower浏览器信息
    print("browser----------")
    print(result.browser.family)
    print(result.browser.version)
    print(result.browser.version_string)

    # device设备信息
    print("device----------")
    print(result.device.family)
    print(result.device.model)

2.3 完整代码

需要将结果数据同时写入到Kafka和HDFS。清洗后的日志，可以用于其他业务分析，具有一定的价值。因为Kafka不能永久保存数据，所以需要把数据存储到HDFS一份。

因为每天都有很多日志，所以需要对日志进行分区。可以通过partitionBy()方法进行分区写入到HDFS。分区的字段需要进行计算。

另外，为了减少小文件生成，可以使用trigger来指定写入的时间间隔。

• 先创建Kafka的Topic

cd /export/server/kafka/bin
./kafka-topics.sh --create --bootstrap-server up01:9092 --topic dwd_nginx_etl_result

• 完整代码

from pyspark.sql import SparkSession
import os
import pyspark.sql.functions as F
from pyspark.sql.types import StringType, MapType
import requests
from user_agents import parse

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

if __name__ == '__main__':
    # 1- 创建SparkSession对象
    spark = SparkSession.builder\
        .appName("nginx_etl")\
        .master("local[*]")\
        .config("spark.sql.shuffle.partitions",2)\
        .getOrCreate()

    # 配置checkpointLocation路径，推荐使用HDFS路径
    spark.conf.set("spark.sql.streaming.checkpointLocation", "hdfs://192.168.88.166:8020/xtzg/chk")

    # 2- 数据输入：读取Kafka中的数据
    """
        "startingOffsets","earliest"：该配置，在实际工作中一般不需要配置。这里是为了开发代码方便
    """
    init_df = spark.readStream.format("kafka")\
        .option("kafka.bootstrap.servers","192.168.88.166:9092")\
        .option("subscribe","xtzg_nginx_log")\
        .option("startingOffsets","earliest")\
        .load()

    # 结构化流中不能以show()方式打印数据数据内容
    # init_df.show()

    # 3- 数据ETL处理
    # 3.1- value字段解码的操作
    """
        cast(StringType())：将字段数据类型强制转换为字符串。等同于SQL语句中的cast(value as string)
        
        下面两种方式都可以，推荐使用第一种，因为性能更好
    """
    # type_cast_df = init_df.select(init_df.value.cast(StringType()).alias("value"))
    type_cast_df = init_df.selectExpr("cast(value as string) as value")

    # 3.2- 通过正则表达式提取Nginx的字段
    pattern = '(?<ip>\d+\.\d+\.\d+\.\d+) (- - \[)(?<datetime>[\s\S]+)(?<t1>\][\s"]+)(?<request>[A-Z]+) (?<url>[\S]*) (?<protocol>[\S]+)["] (?<code>\d+) (?<sendbytes>\d+) ["](?<refferer>[\S]*) ["](?<useragent>[\S\s]+)["] ["](?<proxyaddr>[\S\s]+)["]'

    # 这里不允许使用Python正则表达式，只能使用Java正则表达式
    # pattern = '(?P<ip>.*?) - - \[(?P<time>.*?)\] "(?P<request>.*?)" (?P<status>.*?) (?P<bytes>.*?) "(?P<referer>.*?)" "(?P<ua>.*?)" "(?P<proxy_address>.*)"'
    regexp_df = type_cast_df.select(
        F.regexp_extract("value",pattern,1).alias("ip"),
        F.regexp_extract("value",pattern,3).alias("datetime"),
        F.regexp_extract("value",pattern,4).alias("t1"),
        F.regexp_extract("value",pattern,5).alias("request"),
        F.regexp_extract("value",pattern,6).alias("url"),
        F.regexp_extract("value",pattern,7).alias("protocol"),
        F.regexp_extract("value",pattern,8).alias("code"),
        F.regexp_extract("value",pattern,9).alias("sendbytes"),
        F.regexp_extract("value",pattern,10).alias("refferer"),
        F.regexp_extract("value",pattern,11).alias("useragent"),
        F.regexp_extract("value",pattern,12).alias("proxyaddr")
    )

    # 3.3- 日期时间格式转换
    datetime_df = regexp_df.withColumn("datetime",F.from_unixtime(F.unix_timestamp("datetime","dd/MMM/yyyy:HH:mm:ss Z"),"yyyy-MM-dd HH:mm:ss"))

    # 3.4- IP地理位置解析
    @F.udf(returnType=StringType())
    def parse_ip(ip_str):
        params = {
            "query": ip_str,
            "co": "",
            "resource_id": "6006",
            "oe": "utf8",
        }
        # 发送请求
        response = requests.get(url="https://opendata.baidu.com/api.php", params=params)
        # 解析响应内容
        result = response.json()

        status = result['status']
        if status == '0':
            # 正常
            try:
                return result['data'][0]['location'].split(" ")[0]
            except:
                return "未知区域"
        else:
            return "未知区域"

    area_df = datetime_df.withColumn("area",parse_ip("ip"))

    # 3.5- UA解析
    """
        为什么这里用户自定义函数推荐返回字典？
        方便后续取值
    """
    @F.udf(returnType=MapType(keyType=StringType(),valueType=StringType()))
    def parse_ua(ua_str):
        result = parse(ua_str)
        os = result.os.family
        browser = result.browser.family
        device = result.device.model
        return {"os":os,"browser":browser,"device":device}

    ua_df = area_df.withColumn("os",parse_ua("useragent")['os'])\
        .withColumn("browser", parse_ua("useragent")['browser'])\
        .withColumn("device", parse_ua("useragent")['device'])

    # 4- 数据输出，启动流式任务
    # 4.1- 输出到HDFS
    # 新增一个分区字段
    dt_df = ua_df.withColumn("dt",F.split("datetime"," ")[0])

    # partitionBy表示按照哪个字段进行分区
    dt_df.writeStream.format("orc").partitionBy("dt")\
        .option("path","hdfs://192.168.88.166:8020/xtzg/etl/dwd_nginx_etl_result")\
        .start()

    # 4.2- 输出到Kafka
    # 注意：一般将数据内容转换为JSON格式输出到Kafka中，为了后续使用方便
    # 注意：输出到Kafka中的字段名称只能叫value
    kafka_df = ua_df.select(
        F.to_json(F.struct("ip","datetime","t1","request","url","protocol","code","sendbytes","refferer","useragent","proxyaddr","area","os","browser","device")).alias("value")
    )

    kafka_df.writeStream.format("kafka")\
        .option("kafka.bootstrap.servers","192.168.88.166:9092")\
        .option("topic","dwd_nginx_etl_result")\
        .start()

    # 4.3- 输出到控制台（为了测试）
    # awaitTermination()只能加在最后一个start()的后面
    dt_df.writeStream.format("console").outputMode("append").start().awaitTermination()

可能遇到的错误一：

原因: 结构化流中将数据输出到文件系统中，需要配置checkpointLocation

可能遇到的错误二：

原因: 输出到Kafka中的字段名称只能叫value

2.4 使用Hive读取HDFS数据

• 创建表

CREATE external TABLE dwd.dwd_nginx_etl_result (
    ip string,
    `datetime` string,
    t1 string,
    request string,
    url string,
    protocol string,
    code string,
    sendbytes string,
    refferer string,
    useragent string,
    proxyaddr string,
    area string,
    os string,
    browser string,
    device string
)
    COMMENT 'nginx日志'
    PARTITIONED BY (dt string)
    STORED AS ORC
    LOCATION '/xtzg/etl/dwd_nginx_etl_result'
    TBLsql ('orc.compress' = 'SNAPPY')
;

• 同步分区

MSCK REPAIR TABLE dwd.dwd_nginx_etl_result;

3、指标统计

• 需求

统计实时请求总数(pv)
统计用户数(uv)
统计用户访问所在区域省(类似抖音的位置显示)
统计用户响应状态码
统计用户使用设备终端信息
统计用户操作系统信息
统计用户首次访问系统的时间
统计用户末次访问系统的时间

ip： 用户访问系统的唯一地址
pv：访问系统的页面次数
uv：访问系统的用户数
area：访问系统用户来自的区域，根据ip解析出地址位置
status_code：访问系统用户请求http协议响应状态码
device_os：设备终端，从ua中提取手机或电脑的系统
device_brand：设备品牌名称，从ua中提取手机或电脑的品牌
browser_name：访问系统用户使用的浏览器名称
first_access_time：用户首次访问系统的时间
last_access_time：用户首次访问系统的时间

• Doris建表语句

使用unique模型。

CREATE DATABASE IF NOT EXISTS log_analysis_db;
CREATE TABLE IF NOT EXISTS log_analysis_db.nginx_log_result
(
    ip varchar(15) comment 'ip地址',
    pv int comment 'pv数',
    uv int comment 'uv数',
    area varchar(50) comment '用户所在区域，根据ip解析',
    status_code varchar(10) comment '请求响应状态码',
    device_os varchar(50) comment '设备系统，从ua中提取手机或电脑使用的系统',
    device_brand varchar(50) comment '，从ua中提取手机或电脑的品牌',
    browser_name varchar(50) comment '电脑和手机，使用浏览器，记录浏览器简称',
    first_access_time datetime comment 'nginx日志记录首次访问时间',
    last_access_time datetime comment 'nginx日志记录末次访问时间'
)
UNIQUE KEY(ip)
DISTRIBUTED BY HASH(ip) BUCKETS 10
sql("replication_num" = "1");

• 完整代码

from pyspark.sql import SparkSession, DataFrame
import os
import pyspark.sql.functions as F
from pyspark.sql.types import StringType

os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/bin/python3'

if __name__ == '__main__':
    # 1- 创建SparkSession对象
    spark = SparkSession.builder \
        .appName("nginx_analysis") \
        .master("local[*]") \
        .config("spark.sql.shuffle.partitions", 2) \
        .getOrCreate()

    # 2- 数据输入：读取Kafka中的数据
    init_df = spark.readStream.format("kafka") \
        .option("kafka.bootstrap.servers", "192.168.88.166:9092") \
        .option("subscribe", "dwd_nginx_etl_result") \
        .option("startingOffsets", "earliest") \
        .load()

    # 3- 数据处理
    # 3.1- value字段类型转换
    type_cast_df = init_df.select(init_df.value.cast(StringType()).alias("value"))

    # 3.2- 从JSON中提取一个个字段
    """
        json_tuple与get_json_object的区别
            get_json_object：
                优点：同时能够解析嵌套的JSON
                缺点：一次只能得到一个字段
            json_tuple：
                优点：一次能得到多个字段
                缺点：针对嵌套JSON，只能一层层解析
    """
    parse_json_df = type_cast_df.select(
        F.json_tuple("value","ip","datetime","code","area","os","browser","device")\
            .alias("ip","datetime","status_code","area","device_os","browser_name","device_brand")
    )

    # 3.3- 指标统计
    # F.lit(1)生成一列，每行的数据内容一样，全都是1。与F.col函数作用类似
    # 因为类似area的这些字段的数据类型是字符串，聚合函数没有太适合的，因此使用first
    result_df = parse_json_df.groupBy("ip").agg(
        F.count("ip").alias("pv"),
        F.lit(1).alias("uv"),
        F.first("area").alias("area"),
        F.first("status_code").alias("status_code"),
        F.first("device_os").alias("device_os"),
        F.first("device_brand").alias("device_brand"),
        F.first("browser_name").alias("browser_name"),
        F.min("datetime").alias("first_access_time"),
        F.max("datetime").alias("last_access_time")
    )

    # 4- 数据输出
    # 4.1- 输出到Doris
    def write_2_doris(batch_df:DataFrame, batch_id):
        """
            将DataFrame输出到Doris中
        :param batch_df: 有界的DataFrame
        :param batch_id: 批次ID
        :return:
        """
        # 注意：一般先用append。如果明确知道要怎么做，那可以再使用overwrite
        batch_df.write.jdbc(
            url="jdbc:mysql://192.168.88.166:9030/log_analysis_db?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&useSSL=false",
            table="nginx_log_result",
            mode="append",
            sql={ 'user' : 'root', 'password' : '123456' }
        )


    result_df.writeStream.foreachBatch(write_2_doris).outputMode("update").start()

    # 4.2- 输出到控制台
    result_df.writeStream.format("console").outputMode("update").start().awaitTermination()

• 结果数据核对

./kafka-console-producer.sh --broker-list up01:9092 --topic dwd_nginx_etl_result

{"ip":"210.27.147.62","cookie":"- - [","datetime":"2024-11-14 11:11:11","t1":"] \"","request":"GET","url":"/search.html","protocol":"HTTP/1.1","code":"401","sendbytes":"58840","refferer":"https://www.douyin.com/goods-recommend/search.html?keyword=美味\"","useragent":"Mozilla/5.0 (Linux; U; Android 9; zh-CN; MI 9 Build/PKQ1.181121.001) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/78.0.3904.108 UCBrowser/13.1.6.1096 Mobile Safari/537.36","proxyaddr":"-","area":"广东省广州市","os":"Android","browser":"UC Browser","device":"XiaoMi MI 9","dt":"2024-11-12"}

1、尝试进行用户行为日志的数据ETL、指标统计

提示：核心是如何解析JSON格式，得到一个个独立的字段