데이터 카탈로그

# DataHub을 활용한 데이터 카탈로그 메타데이터 등록
from datahub.emitter.mce_builder import (
    make_dataset_urn, make_tag_urn, make_term_urn,
    make_user_urn, make_domain_urn
)
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import (
    DatasetPropertiesClass, SchemaMetadataClass,
    SchemaFieldClass, StringTypeClass, NumberTypeClass,
    OwnershipClass, OwnerClass, OwnershipTypeClass,
    GlobalTagsClass, TagAssociationClass,
    GlossaryTermsClass, GlossaryTermAssociationClass,
    DomainsClass, StatusClass
)
from datahub.emitter.mcp import MetadataChangeProposalWrapper

# DataHub GMS 서버 연결
emitter = DatahubRestEmitter(
    gms_server="http://datahub-gms:8080",
    token="your-access-token"
)

# 1. 데이터셋 URN 생성 (고유 식별자)
dataset_urn = make_dataset_urn(
    platform="snowflake",
    name="analytics.gold.customer_360",
    env="PROD"
)

# 2. 데이터셋 속성 정의 (비즈니스 메타데이터)
properties = DatasetPropertiesClass(
    name="customer_360",
    description="""
    통합 고객 360도 뷰 테이블.
    주문, 행동, CRM 데이터를 결합한 분석용 마트.

    사용 사례:
    - 고객 세그먼테이션
    - LTV 예측 모델 학습
    - 마케팅 캠페인 타겟팅
    """,
    qualifiedName="analytics.gold.customer_360",
    customProperties={
        "refresh_frequency": "hourly",
        "data_owner": "analytics-team",
        "pii_classification": "confidential",
        "row_count": "15,000,000",
        "data_quality_score": "95%"
    }
)

# 3. 스키마 정의 (기술적 메타데이터)
schema = SchemaMetadataClass(
    schemaName="customer_360",
    platform=make_dataset_urn("snowflake", "", "PROD"),
    version=0,
    hash="",
    platformSchema={},
    fields=[
        SchemaFieldClass(
            fieldPath="customer_id",
            type=StringTypeClass(),
            nativeDataType="VARCHAR(50)",
            description="고객 고유 식별자 (UUID)",
            nullable=False,
            isPartOfKey=True
        ),
        SchemaFieldClass(
            fieldPath="email",
            type=StringTypeClass(),
            nativeDataType="VARCHAR(255)",
            description="고객 이메일 주소 (개인정보)",
            nullable=True,
            globalTags=GlobalTagsClass(tags=[
                TagAssociationClass(tag=make_tag_urn("PII")),
                TagAssociationClass(tag=make_tag_urn("Sensitive"))
            ])
        ),
        SchemaFieldClass(
            fieldPath="lifetime_value",
            type=NumberTypeClass(),
            nativeDataType="DECIMAL(18,2)",
            description="고객 생애 가치 (LTV, USD)",
            nullable=True
        ),
        SchemaFieldClass(
            fieldPath="segment",
            type=StringTypeClass(),
            nativeDataType="VARCHAR(20)",
            description="고객 세그먼트 (VIP/Standard/Basic)",
            nullable=False
        ),
        SchemaFieldClass(
            fieldPath="last_activity_date",
            type=StringTypeClass(),
            nativeDataType="DATE",
            description="마지막 활동 일자",
            nullable=True
        )
    ]
)

# 4. 소유권 설정
ownership = OwnershipClass(
    owners=[
        OwnerClass(
            owner=make_user_urn("analytics_lead"),
            type=OwnershipTypeClass.DATAOWNER
        ),
        OwnerClass(
            owner=make_user_urn("data_steward"),
            type=OwnershipTypeClass.BUSINESS_OWNER
        )
    ]
)

# 5. 비즈니스 용어집(Glossary) 연결
glossary_terms = GlossaryTermsClass(
    terms=[
        GlossaryTermAssociationClass(
            urn=make_term_urn("Customer Analytics")
        ),
        GlossaryTermAssociationClass(
            urn=make_term_urn("Marketing Data")
        )
    ],
    auditStamp={"time": 0, "actor": "urn:li:corpuser:admin"}
)

# 6. 도메인 할당
domains = DomainsClass(
    domains=[make_domain_urn("Marketing")]
)

# 7. 태그 추가
global_tags = GlobalTagsClass(
    tags=[
        TagAssociationClass(tag=make_tag_urn("Gold")),
        TagAssociationClass(tag=make_tag_urn("Certified")),
        TagAssociationClass(tag=make_tag_urn("Production"))
    ]
)

# 8. 메타데이터 발행
mcps = [
    MetadataChangeProposalWrapper(entityUrn=dataset_urn, aspect=properties),
    MetadataChangeProposalWrapper(entityUrn=dataset_urn, aspect=schema),
    MetadataChangeProposalWrapper(entityUrn=dataset_urn, aspect=ownership),
    MetadataChangeProposalWrapper(entityUrn=dataset_urn, aspect=glossary_terms),
    MetadataChangeProposalWrapper(entityUrn=dataset_urn, aspect=domains),
    MetadataChangeProposalWrapper(entityUrn=dataset_urn, aspect=global_tags),
]

for mcp in mcps:
    emitter.emit(mcp)

print(f"Dataset registered: {dataset_urn}")

# 9. GraphQL로 메타데이터 검색
import requests

query = """
query {
    search(input: {
        type: DATASET,
        query: "customer",
        start: 0,
        count: 10
    }) {
        searchResults {
            entity {
                urn
                ... on Dataset {
                    name
                    description
                    platform { name }
                }
            }
        }
    }
}
"""

response = requests.post(
    "http://datahub-gms:8080/api/graphql",
    json={"query": query},
    headers={"Authorization": "Bearer your-token"}
)
print(response.json())