JSON日志处理 | 基于SparkSql实现

标签：purchase users json cart JSON SparkSql 日志 id

0 主要JSON处理函数

get_json_object: 提取单个 JSON 字段
json_tuple: 同时提取多个 JSON 字段
from_json: JSON 字符串转结构化数据
to_json: 结构化数据转 JSON 字符串
schema_of_json: 推断 JSON schema

1 JSON处理函数使用

get_json_object函数允许我们使用JSONPath表达式从JSON字符串中提取特定字段。

语法:

get_json_object(json_str, path)

例:

SELECT get_json_object('{"name":"John", "age":30}', '$.name') AS name;
-- 输出: John

这个函数特别适合从复杂JSON中提取单个字段。

json_tuple: 多字段提取神器

当需要同时提取多个JSON字段时,json_tuple函数是你的最佳选择。

语法:

json_tuple(json_str, key1, key2, ...)

示例:

SELECT json_tuple('{"name":"John", "age":30, "city":"New York"}', 'name', 'age') AS (name, age);
-- 输出: John, 30

json_tuple能显著提高多字段提取的效率,减少重复解析。

to_json: 结构化数据转JSON的便捷工具

与from_json相反,to_json函数将结构化数据转换回JSON字符串。

语法:

to_json(expr[, options])

示例:

SELECT to_json(struct("John" AS name, 30 AS age)) AS json_data;
-- 输出: {"name":"John","age":30}

在数据导出或API响应生成时,这个函数尤为实用。

schema_of_json: JSON Schema推断神器

schema_of_json函数能自动推断JSON字符串的schema,省去手动定义的麻烦。

语法:

schema_of_json(json_str)

示例:

SELECT schema_of_json('{"name":"John", "age":30, "scores":[85, 90, 92]}') AS json_schema;

这个函数在处理未知结构的JSON数据时特别有价值。

JSON数组处理:size和explode函数

处理JSON数组是一个常见需求,SparkSQL为此提供了强大的支持。

size函数:获取数组长度
size函数可以用来获取JSON数组的长度。

语法:

size(json_array)

示例:

SELECT size(from_json('{"scores":[85, 90, 92]}', 'struct<scores:array<int>>').scores) AS array_size;
-- 输出: 3

explode函数:展开JSON数组
explode函数能将JSON数组展开为多行,方便进行后续分析。

语法:

explode(array)

示例:

SELECT explode(from_json('{"scores":[85, 90, 92]}', 'struct<scores:array<int>>').scores) AS score;
-- 输出:
-- 85
-- 90
-- 92

2 案例分析

JSON函数实战:电商用户行为分析

数据样例

{
  "user_id": 1001,
  "session_id": "a1b2c3d4",
  "timestamp": "2024-08-01T10:30:00Z",
  "events": [
    {"type": "view", "product_id": "P001", "category": "Electronics"},
    {"type": "add_to_cart", "product_id": "P001", "quantity": 1},
    {"type": "purchase", "product_id": "P001", "price": 599.99}
  ]
}

数据处理和分析

-- 1. 提取用户ID和会话ID
WITH parsed_logs AS (
  SELECT
    get_json_object(value, '$.user_id') AS user_id,
    get_json_object(value, '$.session_id') AS session_id,
    get_json_object(value, '$.timestamp') AS event_time,
    explode(from_json(get_json_object(value, '$.events'), 'array<struct<type:string,product_id:string,category:string,quantity:int,price:double>>')) AS event
  FROM user_logs
),

-- 2. 分析用户行为
user_behavior AS (
  SELECT
    user_id,
    session_id,
    COUNT(CASE WHEN event.type = 'view' THEN 1 END) AS view_count,
    COUNT(CASE WHEN event.type = 'add_to_cart' THEN 1 END) AS cart_add_count,
    COUNT(CASE WHEN event.type = 'purchase' THEN 1 END) AS purchase_count,
    SUM(CASE WHEN event.type = 'purchase' THEN event.price ELSE 0 END) AS total_purchase_amount
  FROM parsed_logs
  GROUP BY user_id, session_id
),

-- 3. 计算转化率
conversion_rates AS (
  SELECT
    COUNT(DISTINCT CASE WHEN view_count > 0 THEN user_id END) AS users_with_views,
    COUNT(DISTINCT CASE WHEN cart_add_count > 0 THEN user_id END) AS users_with_cart_adds,
    COUNT(DISTINCT CASE WHEN purchase_count > 0 THEN user_id END) AS users_with_purchases
  FROM user_behavior
)

-- 4. 输出分析结果
SELECT
  users_with_views AS total_active_users,
  users_with_cart_adds AS users_adding_to_cart,
  users_with_purchases AS users_making_purchase,
  ROUND(users_with_cart_adds / users_with_views * 100, 2) AS view_to_cart_rate,
  ROUND(users_with_purchases / users_with_cart_adds * 100, 2) AS cart_to_purchase_rate,
  ROUND(users_with_purchases / users_with_views * 100, 2) AS overall_conversion_rate
FROM conversion_rates;

解释:

parsed_logs: 使用get_json_object提取顶层字段,并用explode和from_json展开嵌套的事件数组。
user_behavior: 统计每个用户会话的各类行为次数和总购买金额。
conversion_rates: 计算不同行为的用户数量。

输出结果如下：

+------------------+---------------------+----------------------+-----------------+----------------------+------------------------+
|total_active_users|users_adding_to_cart|users_making_purchase|view_to_cart_rate|cart_to_purchase_rate|overall_conversion_rate|
+------------------+---------------------+----------------------+-----------------+----------------------+------------------------+
|             10000|                6000|                 3000|            60.00|                50.00|                  30.00|
+------------------+---------------------+----------------------+-----------------+----------------------+------------------------+

进一步分析

我们还可以深入分析最受欢迎的产品类别:

SELECT
  event.category,
  COUNT(*) AS view_count,
  SUM(CASE WHEN event.type = 'purchase' THEN 1 ELSE 0 END) AS purchase_count,
  ROUND(SUM(CASE WHEN event.type = 'purchase' THEN 1 ELSE 0 END) / COUNT(*) * 100, 2) AS conversion_rate
FROM parsed_logs
WHERE event.category IS NOT NULL
GROUP BY event.category
ORDER BY view_count DESC
LIMIT 5;

3 小结

在大数据时代，JSON 格式因其灵活性和广泛应用而成为数据处理的重要一环。SparkSQL 提供了强大的内置 JSON 函数，让我们能够高效地处理复杂的 JSON 数据。本文全面总结了这些函数的使用方法、优化技巧及实战应用。

本文核心要点

灵活运用函数组合：如 get_json_object 与 explode 配合使用
性能优先：合理使用 schema 定义，避免过度依赖自动推断
数据层次化处理：使用 CTE (Common Table Expression) 使查询更清晰
商业洞察导向：从原始数据中提取有价值的业务指标

如果您觉得本文还不错，对你有帮助，那么不妨可以关注一下我的数字化建设实践之路专栏，这里的内容会更精彩。

专栏原价99，现在活动价59.9，按照阶梯式增长，还差5个人上升到69.9，最终恢复到原价。

专栏优势：

（1）一次收费持续更新。

（2）实战中总结的SQL技巧，帮助SQLBOY 在SQL语言上有质的飞越，无论你应对业务难题及面试都会游刃有余【全网唯一讲SQL实战技巧，方法独特】

（3）实战中数仓建模技巧总结，让你认识不一样的数仓。【数据建模+业务建模，不一样的认知体系】（如果只懂数据建模而不懂业务建模，数仓体系认知是不全面的）

（4）数字化建设当中遇到难题解决思路及问题思考。

我的专栏具体链接如下：

数字化建设通关指南_莫叫石榴姐的博客-CSDN博客

https://blog.csdn.net/godlovedaniel/category_12706766.html?fromshare=blogcolumn&sharetype=blogcolumn&sharerId=12706766&sharerefer=PC&sharesource=godlovedaniel&sharefrom=from_link

标签：purchase,users,json,cart,JSON,SparkSql,日志,id
From： https://blog.csdn.net/godlovedaniel/article/details/143170926

JSON日志处理 | 基于SparkSql实现

0 主要JSON处理函数

1 JSON处理函数使用

2 案例分析

3 小结

相关文章

赞助商

阅读排行

JSON日志处理 | 基于SparkSql实现

0 主要JSON处理 函数

1 JSON处理函数使用

2 案例分析

3 小结

相关文章

赞助商

阅读排行

0 主要JSON处理函数