使用 Amazon DataZone 和 Open Lineage 实现数据血缘

数据血缘（Data Lineage）描述了数据在整个生命周期中的流转路径和变化过程。它能够追溯数据的来源，跟踪数据在不同系统和流程中的变更历史，以及最终输出的去向。数据血缘包含以下几个关键方面：

数据来源：记录数据的初始输入来源，可能是数据库表、文件、外部系统等
数据转换：描述数据在传输和处理过程中所经历的各种转换操作，如连接、过滤、聚合、清洗等
数据流动：展示数据在不同系统和应用程序之间的流动路径
数据依赖关系：揭示数据对象之间的相互依赖关系，例如表与表、字段与字段之间的关联
数据质量元数据：记录与数据质量相关的元数据，如数据规则、异常情况等
数据输出：标识数据的最终落地位置和使用目的地

通过数据血缘，数据管理人员和消费者能够全面了解数据的“生命旅程”，追根溯源发现问题根源，评估数据质量和可信度，并进行有效的影响分析和审计。这对于确保数据治理、合规性和决策质量至关重要。

在 Amazon DataZone 中，已经实现了与 OpenLineage 兼容的数据血缘功能，帮助我们捕获、存储和可视化数据移动时产生的血缘。关于 OpenLineage 详细的介绍，可以参考 Amazon DataZone 推出 OpenLineage 兼容的数据血缘可视化功能预览版这篇博客。在本篇文章中，我们将详细地介绍如何在 Amazon DataZone 使用 OpenLineage 捕获和展示数据血缘，并详细讨论上述博客中提到的几个例子。

先来看一下本文用到的两个主要的产品。

Amazon DataZone
Amazon DataZone 是一项数据管理服务，可让客户更快、更轻松地对存储在 AWS、本地和第三方来源的数据进行分类、发现、共享和管理。借助 Amazon DataZone，工程师、数据科学家、产品经理、分析师和业务用户可以更轻松地访问整个组织的数据，从而发现、使用和协作得出数据驱动的见解。

OpenLineage
OpenLineage 是用于数据血缘收集和分析的开源框架，是新一代强大的、有上下文意识的数据工具和最佳实践的基础。OpenLineage 能够一致地收集线索元数据，从而更深入地了解数据是如何产生和使用的。目前 Amazon DataZone 已经集成了 OpenLineage 兼容 API, 可以像访问 OpenLineage 一样把数据血缘收集到 Amazon DataZone。

1. 预先准备

在整个实验过程中，我们需要使用到如下资源：

An AWS account
An AWS Identity and Access Management (IAM) user with access to the following services:
AWS Cloud9
Amazon DataZone
AWS Glue
Amazon S3
Amazon Redshift Serverless
Amazon MWAA

在整个实验的过程中，我们会使用 Cloud9 来执行脚本，可以参考文档部署 Cloud9。

2. DataZone

我们先来准备 DataZone 的运行环境。流程图如下：

2.1 Domain

DataZone Domain 是用于连接我们的数据资产、用户及其项目的组织实体，通过 Domain, 我们可以灵活地反映组织结构的数据和分析需求。我们可以为整个企业创建单个 Domain, 也可以为不同的业务部门或团队创建各自的 Domain. 我们可以参考文档来创建一个 Domain。

2.2 Blueprints

创建好 Domain 后，必须启用 Blueprints，它定义了在 DataZone 环境里可以使用的工具和服务，例如 Glue。我们把如下图所示的 4 个 Blueprints 都启用。

2.3 Project

Project 使一组用户能够在各种涉及发布、发现、订阅和消费 DataZone 目录中数据资产的业务用例上进行协作。我们可以参考文档来创建 Project。

2.4 Environment Profile

Environment Profile 是一种模板，我们可以使用它来创建 Environment，通过在 Environment Profile 中嵌入诸如 AWS 账户和区域等信息，从而简化 Environment 的创建。我们可以参考文档来创建 Environment Profile。

创建 Environment Profile 的时候，需要选定 Blueprint，不同的 Blueprint 代表了 AWS 不同的服务，我们分别使用 “Default Data Lake” 和 “Default Data Warehouse” 这两个 Blueprint 创建两个 Environment Profile。

在使用 “Default Data Warehouse” 创建 Environment Profile 的时候，需要先创建 Amazon Redshift 集群或者 Serverless 部署模式，可以参考文档来创建或者部署；并参考文档来创建访问 Redshift 的 AWS Secret Manager 对象。

2.5 Environment

在 Amazon DataZone 项目中，Environment 是已配置资源的集合（例如 Amazon S3 存储桶、AWS Glue 数据库或 Amazon Athena 工作组），具有指定的一组 IAM 主体，即 Environment 用户的角色，这些角色被分配了所有者或贡献者权限，可以对这些资源进行操作。

我们使用上一步创建的两种 Environment Profile，分别创建两个 Environment。创建 Data Lake 类型的 Environment 的时候，会自动创建用于发布和订阅数据资产的 Glue 数据库，可以指定名称或者自动命名。请注意：如果使用了 Lake Formation，请确认当前用户已经是 “Data lake administrators”。

创建 Data Warehouse 类型的 Environment 的时候，会自动创建用于发布和订阅 Redshift Schema，可以指定名称或者自动命名。请注意：如果使用了 AWS Secret 存储访问 Redshift的用户名和密码，请一定为 AWS Secret 打上AmazonDataZoneProject 和 AmazonDataZoneDomain 这两个 Tag, 否则就会出现 Environment 因为访问不到 AWS Secret 而导致创建失败。

创建 Data Lake 类型的 Environment 的时候，会同步创建默认的 Data Source, 在后面例子中收集的数据血缘，都在 Data Source 中查看。如果需要在 Environment 中新增 Data Source，请参考文档。

3. Glue Crawler 的例子

先看一个捕获、收集和展示 Glue Crawler 的数据血缘的例子。

3.1 准备 S3

我们可以在 Cloud9 上使用如下脚本，来准备测试数据：

ACCOUNT_ID=<Your-Account-ID>
REGION=<Your-Reginon>
aws s3 mb s3://"datazone-$ACCOUNT_ID" --region $REGION

aws s3api put-object --bucket "datazone-$ACCOUNT_ID" --key inventory/
wget https://aws-blogs-artifacts-public.s3.amazonaws.com/BDB-4447/data/inventory.csv
aws s3 cp inventory.csv s3://"datazone-$ACCOUNT_ID"/inventory/

aws s3api put-object --bucket "datazone-$ACCOUNT_ID" --key scripts/
wget https://aws-blogs-artifacts-public.s3.amazonaws.com/BDB-4447/scripts/Inventory_Insights.py
aws s3 cp Inventory_Insights.py s3://"datazone-$ACCOUNT_ID"/scripts/

aws s3api put-object --bucket "datazone-$ACCOUNT_ID" --key lib/
wget https://aws-blogs-artifacts-public.s3.amazonaws.com/BDB-4447/lib/openlineage-spark_2.12-1.9.1.jar
aws s3 cp openlineage-spark_2.12-1.9.1.jar s3://"datazone-$ACCOUNT_ID"/lib/

请手动替换尖括号里的内容，以后不再赘述。

3.3 Glue Crawler

使用如下脚本，准备 Glue Crawler 的配置文件 crawler.json 如下：

{
    "Name": "data-lineage-demo",
    "Role": "arn:aws:iam::<Your-Account-ID>:role/<Glue-Crawler-Role>",
    "Targets": {
        "S3Targets": [
            {
                "Path": "s3://datazone-<Your-Account-ID>/inventory/"
            },
            {
                "Path": "s3://datazone-<Your-Account-ID>/inventory_insights/"
            }
        ]
    },
    "DatabaseName": "<Database-Name>"
}

替换上面脚本中的 Database-Name 为在创建 DataZone Environment 时创建的用于发布的数据库。使用如下脚本，创建 Glue Crawler：

aws glue create-crawler --cli-input-json file://crawler.json

如果使用了 Lake Formation，需要对 Crawler 的执行权限进行授权，使其具有对数据库中建表的权限。可以使用如下脚本实现：

aws lakeformation grant-permissions \
  --principal DataLakePrincipalIdentifier=arn:aws:iam::<Your-Account-ID>:role/<Glue-Crawler-Role> \
  --resource '{"Database": {"Name": "<Database-Name>"}}' \
  --permissions ALL \
  --permissions-with-grant-option ALL \
  --region us-east-1

运行 Glue Crawler，将上面准备的 S3 的数据，爬取表结构到 Glue 数据库中，也就是我们在创建 DataZone Environment 时创建的数据库：

aws glue start-crawler --name data-lineage-demo

3.4 收集 Glue Crawler 的血缘

我们先下载收集 Glue Crawler 血缘的代码 extract_glue_crawler_lineage.py，并上传到 Cloud9，在 Cloud9 上执行如下代码：

python3 -m venv env
. env/bin/activate
pip install boto3

# 在 DataZone 控制台或者 Portal URL 地址上可以看到到 Domain 的 ID
python extract_glue_crawler_lineage.py -d <Database-Name> -t inventory -r <Your-Reginon> -i <Domain->

该 Python 脚本会解析 Glue Crawler 执行成功的日志，并将日志中的血缘信息写到 DataZone。请注意：脚本里 matched_crawl 使用了第一条事件，如果有多次事件，可以修改脚本，在获取 matched_crawls 后，按结束时间降序排序，取最新的事件。

matched_crawls = [
        crawl
        for crawl in crawls["Crawls"]
        if (
            crawl["State"] == "COMPLETED"
            and crawl["EndTime"] >= table["CreateTime"]
            and crawl["StartTime"] <= table["CreateTime"]
        )
     ]
# 按结束时间降序排序，取最新的事件
matched_crawl = sorted(matched_crawls, key=lambda x: x["EndTime"], reverse=True)[0]

3.5 展示血缘

使用上一步的脚本，将 Glue Crawler 成功运行的事件收集到 DataZone，登录我们在前面章节创建的 DataZone，并进入创建的 Domain 和 Project，进入上一章节创建的 Data Lake 类型的 Environment 默认创建的 Data Source，点击右上角的 “RUN” 按钮来运行它，如下图：

运行完成后，会显示 Glue Crawler 生成的表，点击进入后，在 “LINEAGE” 这个栏目下，可以看到这个表的血缘，如下图：

至此，Glue Crawler 生成表的的血缘的捕获和展示，已经演示完成了。

4. Glue ETL 的例子

我们再来看一个捕获、收集和展示 Glue ETL 的血缘的例子。

4.1 创建 Glue ETL Job

在 Cloud9 使用 AWS CLI 命令创建 Glue ETL Job. 先创建一个 Python 脚本 Inventory_Insights.py，内容如下：

import sys
import boto3
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import col

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

# Fetch AWS account number
sts_client = boto3.client('sts')
account_id = sts_client.get_caller_identity()['Account']

# Load data from the Glue Catalog as a DataFrame
df = spark.sql("SELECT * FROM <Database-Name>.inventory")

# Log initial DataFrame count
print(f"Initial DataFrame count: {df.count()}")

# Show schema to verify column names and data types
df.printSchema()

# Show sample data
df.show(5)

# Group by product_id and sum the inventory_quantity
grouped_df = df.groupBy("product_id").sum("inventory_quantity").withColumnRenamed("sum(inventory_quantity)", "total_inventory")
print(f"Grouped DataFrame count: {grouped_df.count()}")

# Show sample data from the grouped DataFrame
grouped_df.show()

# Construct the S3 path including the account number
s3_path = f"s3://datazone-{account_id}/inventory_insights/"

# Write the data out in Parquet format to S3, overwriting existing data
grouped_df.write.mode("overwrite").format("parquet").save(s3_path)

job.commit()

请替换上面脚本中的<Database-Name> 为上一个例子中，Glue Crawler 用到的 Database，该脚本实现了对表 inventory 的聚合，并将结果写到 S3 的指定目录。将 Python 脚本上传到 S3：

aws s3 cp Inventory_Insights.py s3://<S3-Bucket>/scripts

创建执行 Glue ETL Job 的角色：

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "glue.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

aws iam create-role --role-name DZBlogRole --assume-role-policy-document file://trust-policy.json

并为该角色添加 S3 和 Glue 的访问策略。

准备 Glue ETL Job 的配置文件 job_configuration.json，内容如下：

{
    "Name": "inventory_insights",
    "Role": "arn:aws:iam::<Your-Account-ID>:role/DZBlogRole",
    "ExecutionProperty": {
        "MaxConcurrentRuns": 1
    },
    "Command": {
        "Name": "glueetl",
        "ScriptLocation": "s3://datazone-<Your-Account-ID>/scripts/Inventory_Insights.py",
        "PythonVersion": "3"
    },
    "DefaultArguments": {
        "--enable-metrics": "true",
        "--spark-event-logs-path": "s3://aws-glue-assets-<Your-Account-ID>-us-east-1/sparkHistoryLogs/",
        "--enable-job-insights": "false",
        "--enable-observability-metrics": "true",
        "--conf": "spark.extraListeners=io.openlineage.spark.agent.OpenLineageSparkListener --conf spark.openlineage.transport.type=amazon_datazone_api --conf spark.openlineage.transport.domainId=dzd_4kcjyh4pf7om1z --conf spark.glue.accountId=551831295244 --conf spark.openlineage.facets.custom_environment_variables=[AWS_DEFAULT_REGION;GLUE_VERSION;GLUE_COMMAND_CRITERIA;GLUE_PYTHON_VERSION;]",
        "--enable-glue-datacatalog": "true",
        "--enable-continuous-cloudwatch-log": "true",
        "--job-bookmark-option": "job-bookmark-disable",
        "--job-language": "python",
        "--TempDir": "s3://aws-glue-assets-<Your-Account-ID>-us-east-1/temporary/"
    },
    "MaxRetries": 0,
    "Timeout": 60,
    "WorkerType": "G.1X",
    "NumberOfWorkers": 2,
    "GlueVersion": "5.0"
}

使用 AWS CLI 创建 Glue ETL job：

aws glue create-job --cli-input-json file://job_configuration.json

4.2 收集血缘

手动执行这个 Glue ETL Job，请注意执行角色 DZBlogRole 需要在 Lake Formation 里具备对表 inventory 的访问权限。这个Glue ETL Job 完成后，它的血缘已经被收集到 Data Zone.

运行 Glue Crawler，将 Glue ETL Job 生成的 S3 上的数据，爬取表结构到 Glue 数据库中，也就是我们在创建 DataZone Environment 时创建的数据库：

aws glue start-crawler --name data-lineage-demo

将 Glue Crawler 成功运行的事件收集到 DataZone，脚本如下：

python3 -m venv env
. env/bin/activate

python extract_glue_crawler_lineage.py -d <Database-Name> -t inventory_insight -r <Your-Region> -i <Domain-ID>

4.3 展示血缘

类似 Glue Crawler 的例子中，登录我们在前面章节创建的 DataZone，并进入创建的 Domain 和 Project，进入上一章节创建的 Data Lake 类型的 Environment 默认创建的 Data Source，点击右上角的 “RUN” 按钮来运行它，如下图：

运行完成后，会显示 Glue Crawler 生成的表 inventory_insights，点击进入后，在 “LINEAGE” 这个栏目下，可以看到这个表的血缘，如下图：

可以点击图中的 “<<” 按钮，展示 Glue ETL Job 产生的血缘。至此，Glue ETL Job 生成的血缘的捕获和展示，已经演示完成了。

5. Redshift 的例子

再来看一个捕获、收集和展示 Redshift SQL 的数据血缘的例子。

5.1 在 Redshift 执行 SQL

在前面章节中创建 Environment 的时候，我们已经部署好了 Redshift Serverless, 并且在创建 Data Warehouse 类型的 Environment 的时候，自动在 Redshift 上创建了 Schema，假定这个 Schema 的名称是 makert, 我们在 Redshift Query Editor 执行如下 SQL：

create table market.retail_sales (
  id BIGINT primary key,
  name character varying not null
);

create table market.online_sales (
  id BIGINT primary key,
  name character varying not null
);

INSERT INTO market.retail_sales VALUES (123, 'item1');
INSERT INTO market.online_sales VALUES (234, 'item2');

create table market.sales AS
Select id, name from market.retail_sales Union ALL
Select id, name from market.online_sales;

在上面的 SQL 中，我们创建了两个表，写入测试数据，并用这两个表生成了一个新的表。

5.2 收集血缘

在 Cloud9 上下载依赖包文件 requirements.txt 和收集 Redshift SQL 血缘的脚本 extract_redshift_lineage.py。

先来安装依赖包，执行脚本如下：

python3 -m venv env 
. env/bin/activate

pip install -r requirements.txt

再来收集血缘，执行脚本如下：

python extract_redshift_lineage.py \
 -r <Your-Region> \
 -i <> \
 -n <Redshift-Serverless-WorkGroup> \
 -t <Redshift-Serverless-Port> \
 -d <Redshift-Serverless-Database> \
 -s <Date> -u <Redshift-Serverless-User> -w <Redshift-Serverless-Password>

上面的脚本会在 Redshift 中搜索指定时间以后执行的 SQL 语句，并搜索相应的表结构，获取 SQL 执行产生的血缘，发送到 DataZone 指定的 Domain。

5.3 展示血缘

类似前面的例子中，登录我们在前面章节创建的 DataZone 并进入创建的 Domain 和 Project，进入上一章节创建的 Data Warehouse 类型的 Environment 默认创建的 Data Source，点击右上角的 “RUN” 按钮来运行它，如下图：

运行完成后，会显示 Redshift 中运行的 SQL 语句生成的表，点击表 “sales” 进入后，在 “LINEAGE” 这个栏目下，可以看到这个表的血缘，如下图：

可以看出，表 sales 的血缘跟我们前面执行的 SQL 是一一对应的。

6. MWAA 的例子

我们将在 Amazon Managed Workflows for Apache Airflow (Amazon MWAA) 中实现一个数据加工的 ETL 过程，并通过 OpenLineage 从 MWAA 中捕获血缘并导入到 DataZone。

6.1 部署 MWAA

我们可以参考 MWAA 的部署文档，来部署 MWAA 的环境，我们采用的 MWAA 版本是 2.6.3，在部署之前，我们先准备 Open Lineage 需要的 Requirements 和 Plugin。

准备 requirements.txt 文件，内容如下：

openlineage-airflow==1.4.1
openapi-python-client

准备 Plugin 的文件 env_var_plugin.py, 内容如下：

from airflow.plugins_manager import AirflowPlugin
from airflow.models import Variable
import os

os.environ["AIRFLOW__OPENLINEAGE__DISABLED"] = "False"
os.environ["AIRFLOW__OPENLINEAGE__TRANSPORT"] = """{"type": "console"}"""

class EnvVarPlugin(AirflowPlugin):
    name = "env_var_plugin"

将文件打包到 plugin.zip 文件，并与 requirements.txt 一起上传到 S3，可以在部署 MWAA 的时候使用。

6.2 部署 DAG

准备要执行的 DAG 代码，先创建一个 Python 脚本 example_redshift_job.py，内容如下：

import os, re
from datetime import datetime, timedelta
from airflow import DAG
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.operators.empty import EmptyOperator
from airflow.utils.dates import days_ago


with DAG(
    dag_id='example_redshift_job',
    schedule_interval = '@daily',
    start_date = days_ago(1),
    tags=['example'],
    catchup=False
) as dag:

    begin = EmptyOperator(task_id="begin")
    end = EmptyOperator(task_id="end")

    create_source_table_sql = f'''
    create table market.retail_sales (
      id BIGINT primary key,
      name character varying not null
    );
    '''

    create_source_table = PostgresOperator(
        task_id="create_source_table",
        postgres_conn_id="redshift_default",
        sql=create_source_table_sql
    )

    insert_source_table_sql = f'''
    INSERT INTO market.retail_sales VALUES (123, 'item1');
    '''

    insert_source_table = PostgresOperator(
        task_id="insert_source_table",
        postgres_conn_id="redshift_default",
        sql=insert_source_table_sql
    )

    create_target_table_sql = f'''
    create table market.sales as
        Select id, name from market.retail_sales;
    '''

    create_target_table = PostgresOperator(
        task_id="create_target_table",
        postgres_conn_id="redshift_default",
        sql=create_target_table_sql
    )

    begin >> create_source_table >> insert_source_table >> create_target_table >> end

这个脚本执行了一个简单的 Redshift ETL 操作，将上面的 Python 脚本上传到 S3 上 DAG 指定的目录：

aws s3 cp example_redshift_job.py s3://<S3-Bucket>/dags/

6.3 收集血缘

当我们在 MWAA 执行 DAG 的时候，根据我们的配置，会将 DAG 中执行的 SQL 日志输出 CloudWatch，其中就包含了数据血缘的相关信息，我们需要执行一个 Python 代码，将这些数据血缘从 CloudWatch 写入 DataZone。

下载 Python 代码 extract_airflow_lineage.py，并在 Cloud9 中执行如下：

python -m venv env
pip install -r requirements.txt
. env/bin/activate

# run the script
python extract_airflow_lineage.py \
  --region us-east-1 \
  --domain-identifier <Domain-ID> \
  --airflow-environment-name <MWAA-Name>
  
# 这里如果测试时，有DAG任务失败，可以添加时间参数，过滤最新的任务日志
# --start-time 2025-05-12T10:05:00

Python 代码会一直运行，可以在收集血缘之后，通过 Ctrl + C 来终止它。

6.4 执行DAG

在执行 DAG 之前，要先在 MWAA 中创建 Redshift 的访问连接 redshift_default，可以参考 Airflow 相关文档来创建。

执行DAG成功后，可以在捕获血缘的 Python 代码的执行过程中，看到如下的日志：

这表示 DAG 中执行的 SQL 语句的数据血缘，已经被收集到了 DataZone 里。

6.5 展示血缘

类似前面的例子中，登录我们在前面章节创建的 DataZone，并进入创建的 Domain 和 Project，进入上一章节创建的 Data Warehouse 类型的 Environment 默认创建的 Data Source，点击右上角的 “RUN” 按钮来运行它，如下图：

运行完成后，会显示 MWAA 的 DAG 中运行的 SQL 语句生成的表，点击表 “sales” 进入后，在 “LINEAGE” 这个栏目下，可以看到这个表的血缘，如下图：

可以看出，表 sales 的血缘跟我们在 DAG 中运行的 SQL 是一一对应的。

6.6 注意事项

在做这个实验的时候，请注意如下：

版本的问题

这个实验涉及到 MWAA 的版本和 OpenLineage 的版本。目前测试成功的案例使用的 MWAA 版本是 2.6.3，OpenLineage 版本是 1.4.1，其它的版本作者做了一些测试，也遇到了一些问题，用户在使用的时候可以针对具体版本再测试一下。

收集血缘的范围

目前 MWAA 和 OpenLineage 只能收集 Redshift SQL 的血缘，并不能收集 EMR Spark SQL 的血缘。对于很多客户使用的 Redshift Copy 语句产生的血缘，也不能收集。

数据血缘的格式

不同版本的 OpenLineage 生成的数据血缘的格式是不同的，如果没有出现 6.4 中的图里的效果，可能是收集血缘的脚本extract_airflow_lineage.py 没有识别 OpenLineage 生成的数据血缘的格式，可以检查一下这个脚本。

其它的展现形式

除了 DataZone，我们还可以使用开源工具 Marquez 来展现数据血缘，可以参考例子 Automating data lineage on Amazon MWAA with OpenLineage。

7. 资源释放

测试结束后，需要释放的资源包括：

AWS Cloud9
Amazon DataZone
AWS Glue
Amazon S3
Amazon Redshift Serverless
Amazon MWAA

8. 总结

在这篇文章中，我们通过 4 个例子，详细地为大家演示了如何在 Amazon DataZone 使用 OpenLineage 捕获和展示数据血缘。Amazon DataZone 作为企业级数据管理服务，提供了强大的血缘追踪能力和与 AWS 生态系统的无缝集成。而 OpenLineage 作为开源框架，则带来了标准化的元数据模型和广泛的集成选项。通过将两者结合，既可以利用 Amazon DataZone 的企业级功能，又能借助 OpenLineage 的开放性和灵活性来扩展其数据血缘能力。

亚马逊AWS官方博客