SparkSQL中的JSON内置函数全解析

get_json_object(json_str, path)

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

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

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

from_json(json_str, schema[, options])

SELECT from_json('{"name":"John", "age":30}', 'struct<name:string, age:int>') AS parsed_data;

to_json(expr[, options])

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

schema_of_json(json_str)

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

size(json_array)

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

explode(array)

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

-- 将JSON数据保存为Parquet格式
CREATE TABLE parquet_table
USING PARQUET
AS SELECT * FROM json_table;

-- 按日期分区存储JSON数据
CREATE TABLE partitioned_json_table (
  id INT,
  data STRING,
  date STRING
)
USING JSON
PARTITIONED BY (date);

CREATE TABLE parsed_json_table AS
SELECT 
  id,
  get_json_object(data, '$.name') AS name,
  get_json_object(data, '$.age') AS age,
  data
FROM json_table;

{
  "user_id": 1001,
  "timestamp": "2024-08-01T10:30:00Z",
  "actions": [
    {"type": "click", "target": "button1"},
    {"type": "view", "target": "page2"}
  ]
}

WITH parsed_logs AS (
  SELECT
    get_json_object(log, '$.user_id') AS user_id,
    explode(from_json(get_json_object(log, '$.actions'), 'array<struct<type:string,target:string>>')) AS action
  FROM log_table
)
SELECT
  user_id,
  COUNT(*) AS click_count
FROM parsed_logs
WHERE action.type = 'click'
GROUP BY user_id
ORDER BY click_count DESC
LIMIT 10;

{
  "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}
  ]
}

from pyspark.sql import SparkSession
spark = SparkSession.builder \
    .appName("E-commerce User Behavior Analysis") \
    .getOrCreate()

df = spark.read.json("path/to/user_logs.json")
df.createOrReplaceTempView("user_logs")

-- 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;

result = spark.sql("""
  -- 在这里粘贴上面的SQL查询
""")
result.show()

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;

WITH parsed_logs AS (
  SELECT
    get_json_object(value, '$.user_id') AS user_id,
    get_json_object(value, '$.session_id') AS session_id,
    explode(from_json(get_json_object(value, '$.events'), 'array<struct<type:string,...>>')) AS event
  FROM user_logs
),
-- 后续数据处理和分析...