<ifrs:ProfitLossFromContinuingOperations contextRef="2023Q3" unitRef="USD">150000000</ifrs:ProfitLossFromContinuingOperations>

{
  "resourceType": "Patient",
  "id": "patient-123",
  "name": [{"family": "張", "given": ["三"]}],
  "birthDate": "1990-01-01",
  "gender": "male",
  "address": [{"line": ["北京市朝陽區"], "city": "北京"}],
  "extension": [  // 自定義擴展字段（如“過敏史”）
    {
      "url": "http://example.org/allergy",
      "valueString": "青黴素過敏"
    }
  ]
}

from pyatlas.client import AtlasClient

# 初始化Atlas客户端
client = AtlasClient(
    base_url="http://atlas-server:21000",
    username="admin",
    password="admin"
)

# 定義User實體類型（業務元數據）
user_entity_def = {
    "name": "User",
    "description": "電商用户維度表",
    "attributeDefs": [
        {
            "name": "userId",
            "typeName": "string",
            "isRequired": True,
            "cardinality": "SINGLE",
            "description": "用户唯一ID，前綴user_"
        },
        {
            "name": "registerTime",
            "typeName": "date",
            "isRequired": True,
            "cardinality": "SINGLE",
            "description": "註冊時間，格式ISO 8601"
        },
        {
            "name": "gender",
            "typeName": "string",
            "isRequired": False,
            "cardinality": "SINGLE",
            "description": "性別，可選值M/F/Unknown"
        }
    ]
}

# 創建User實體類型
client.create_type(user_entity_def)

# 新增User實例（技術元數據）
user_instance = {
    "typeName": "User",
    "attributes": {
        "userId": "user_123",
        "registerTime": "2023-01-01T00:00:00Z",
        "gender": "M"
    }
}

client.create_entity(user_instance)

import great_expectations as ge
from great_expectations.dataset import SparkDFDataset

# 加載標準化後的數據（Spark DataFrame）
df = spark.read.parquet("hdfs:///user/behavior/standardized")
gdf = SparkDFDataset(df)

# 定義質量校驗規則
expectations = [
    # 1. user_id非空
    gdf.expect_column_not_to_be_null("user_id"),
    # 2. behavior_type只能是view/click/purchase
    gdf.expect_column_values_to_be_in_set("behavior_type", ["view", "click", "purchase"]),
    # 3. event_time在2023年範圍內
    gdf.expect_column_values_to_be_between("event_time", "2023-01-01T00:00:00Z", "2023-12-31T23:59:59Z"),
    # 4. stay_duration≥0（頁面停留時長不能為負）
    gdf.expect_column_values_to_be_greater_than_or_equal_to("stay_duration", 0)
]

# 執行校驗並輸出結果
results = gdf.validate(expectations=expectations)
print(results)

{
  "user_id": 123,  // 數字格式（需標準化為string）
  "event_time": "2023-10-01 12:00:00",  // 非ISO格式（需轉成timestamp）
  "product_id": "prod_456",
  "behavior_type": "Click",  // 大寫（需轉小寫）
  "stay_duration": "10"  // 字符串（需轉int）
}

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lower, to_timestamp

spark = SparkSession.builder.appName("BehaviorStandardization").getOrCreate()

# 讀取原始JSON數據
raw_df = spark.read.json("hdfs:///user/behavior/raw")

# 數據清洗與標準化
standardized_df = raw_df \
    .withColumn("user_id", col("user_id").cast("string"))  # 轉string
    .withColumn("event_time", to_timestamp("event_time", "yyyy-MM-dd HH:mm:ss"))  # 轉ISO timestamp
    .withColumn("behavior_type", lower(col("behavior_type")))  # 轉小寫
    .withColumn("stay_duration", col("stay_duration").cast("int"))  # 轉int
    .select("user_id", "event_time", "product_id", "behavior_type", "stay_duration")  # 保留核心字段

# 存儲為Parquet格式（按event_time分區）
standardized_df.write \
    .mode("overwrite") \
    .partitionBy("event_time") \
    .parquet("hdfs:///user/behavior/standardized")

from pyatlas.client import AtlasClient

client = AtlasClient(base_url="http://atlas-server:21000", username="admin", password="admin")

# 定義Fact_User_Behavior實體類型
behavior_entity = {
    "name": "Fact_User_Behavior",
    "description": "電商用户行為事實表",
    "attributeDefs": [
        {"name": "user_id", "typeName": "string", "isRequired": True},
        {"name": "event_time", "typeName": "timestamp", "isRequired": True},
        {"name": "product_id", "typeName": "string", "isRequired": True},
        {"name": "behavior_type", "typeName": "string", "isRequired": True},
        {"name": "stay_duration", "typeName": "int", "isRequired": False}
    ]
}

# 創建實體類型
client.create_type(behavior_entity)

# 註冊表實例
table_instance = {
    "typeName": "Fact_User_Behavior",
    "attributes": {
        "name": "fact_user_behavior",
        "qualifiedName": "hdfs:///user/behavior/standardized",
        "description": "電商用户行為事實表",
        "owner": "data_engineer"
    }
}

client.create_entity(table_instance)

import great_expectations as ge
from great_expectations.dataset import SparkDFDataset

# 加載標準化數據
df = spark.read.parquet("hdfs:///user/behavior/standardized")
gdf = SparkDFDataset(df)

# 定義校驗規則
results = gdf.validate(expectations=[
    gdf.expect_column_not_to_be_null("user_id"),
    gdf.expect_column_values_to_be_in_set("behavior_type", ["view", "click", "purchase"]),
    gdf.expect_column_values_to_be_greater_than_or_equal_to("stay_duration", 0)
])

# 輸出校驗結果（若有失敗，打印錯誤信息）
if not results["success"]:
    print("數據質量校驗失敗：")
    for result in results["results"]:
        if not result["success"]:
            print(f"- {result['expectation_config']['kwargs']['column']}：{result['result']['message']}")

from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer

# 定義Avro schema
behavior_schema = """
{
  "type": "record",
  "name": "UserBehavior",
  "fields": [
    {"name": "user_id", "type": "string"},
    {"name": "event_time", "type": "string", "logicalType": "timestamp-millis"},
    {"name": "product_id", "type": "string"},
    {"name": "behavior_type", "type": "string"},
    {"name": "stay_duration", "type": "int"}
  ]
}
"""

# 註冊schema到Schema Registry
schema_registry_client = SchemaRegistryClient({"url": "http://schema-registry:8081"})
schema_id = schema_registry_client.register_schema("user_behavior-value", behavior_schema)

# 序列化數據（生產者用）
avro_serializer = AvroSerializer(behavior_schema, schema_registry_client)
data = {"user_id": "user_123", "event_time": "2023-10-01T12:00:00Z", "product_id": "prod_456", "behavior_type": "click", "stay_duration": 10}
serialized_data = avro_serializer(data)

flowchart LR
    A[原始數據採集] --> B[格式標準化（Spark）]
    B --> C[元數據註冊（Atlas）]
    C --> D[質量校驗（Great Expectations）]
    D --> E[存儲（Parquet/Avro）]
    E --> F[應用（分析/集成/合規）]