Apache Airflow 是一个开源平台，用于开发、调度和监控批处理工作流。

ByteHouse 支持通过 IAM 用户或数据库用户连接 Flink Connector Driver。IAM 用户与数据库用户二者差异说明如下，您可按需选择。

• IAM 用户为火山引擎访问控制（IAM）中创建的用户，其权限由 IAM 权限策略及您授予的 ByteHouse 资源和数据权限决定。IAM 用户可访问 ByteHouse 控制台，也支持通过 CLI、连接驱动、生态工具、API 等方式访问 ByteHouse。
• 数据库用户为 ByteHouse 中创建的数据库级别用户，可为其授予环境、资源和数据权限。数据库用户不可访问 ByteHouse 控制台，但支持通过 CLI、连接驱动、生态工具、API 等方式访问 ByteHouse。

更多 IAM 用户和数据库用户的介绍请参见以下文档：

• ByteHouse权限管控概述
• 管理 IAM 用户
• 管理数据库用户

• 在您的虚拟/本地环境中安装 pip。
• 在您的虚拟/本地环境中安装 ByteHouse CLI，并使用您自己的 ByteHouse 账户登录。对于 macOS，您可以直接通过 homebrew 安装它。

  brew install bytehouse-cli
  

首先，在您的本地环境或虚拟环境中安装 Apache Airflow。在本教程中，我们使用 pip 进行安装。有关其他安装方法，请查看 Apache Airflow 官方文档以获取详细信息。

# 使用 Airflow 前，您需要指定一个目录；默认目录是 ~/airflow
# （可选）您也可以按需选择其他路径
export AIRFLOW_HOME=~/airflow

# 执行以下命令
AIRFLOW_VERSION=2.7.0
PYTHON_VERSION="$(python --version | cut -d " " -f 2 | cut -d "." -f 1-2)"

CONSTRAINT_URL="https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-${PYTHON_VERSION}.txt"
pip install "apache-airflow==${AIRFLOW_VERSION}" --constraint "${CONSTRAINT_URL}"

安装完成后，运行命令airflow info以获取有关 Airflow 的更多信息。

通过执行以下命令来初始化 Airflow 的 Web 服务器：

# 初始化数据库
airflow db init

airflow users create \
    --username admin \
    --firstname admin \
    --lastname admin \
    --role Admin \
    --email admin 
# 启动 Web 服务器，默认端口是 8080
# 或者修改 airflow.cfg 文件来设置 web_server_port
airflow webserver --port 8080

设置完 Web 服务器后，您可以访问 http://localhost:8080/，使用之前设置的用户名和密码登录 Airflow 控制台。

在一个新的终端中，通过以下命令设置 Airflow 调度程序，然后刷新 http://localhost:8080/。

# 启动调度器
# 打开一个新终端，或者使用 "-D" 选项运行 web 服务器，将其作为守护进程运行
airflow scheduler
# 在浏览器中访问 localhost:8080，使用您刚刚创建的 admin 账号登录
# 在主页上启用 example_bash_operator dag

1. 在 airflow 的路径下创建名为 dags 的文件夹，然后创建 test_bytehouse.py 以启动新的 DAG 作业。

   ~/airflow
   mkdir dags
   cd dags
   nano test_bytehouse.py
   

2. 在test_bytehouse.py中添加以下代码，该作业可以连接到 ByteHouse CLI 并使用 BashOperator 运行任务，以运行查询或将数据加载到 ByteHouse 中。您可选择使用 IAM 用户或数据库用户连接 ByteHouse。

   使用 IAM 用户

   使用数据库用户

   from datetime import timedelta
      import pendulum
      from airflow import DAG
      from airflow.operators.bash import BashOperator
      default_args = {
          'owner': 'airflow',
          'depends_on_past': False,
          'email': ['airflow@example.com'],
          'email_on_failure': False,
          'email_on_retry': False,
          'retries': 1,
          'retry_delay': timedelta(minutes=1),
      }
      # 创建 DAG 实例
      with DAG(
          'test_bytehouse',
          default_args=default_args,
          description='A simple tutorial DAG',
          schedule=timedelta(days=1),  
          start_date=pendulum.today('UTC').add(days=-1),
          tags=['example'],
      ) as dag:
          # 定义 BashOperators
          tCreateDatabase = BashOperator(
              task_id='ch_database',
              depends_on_past=False,
              bash_command='bytehouse-cli --host {HOST} --port {PORT} --secure --token {API_KEY} -q "CREATE DATABASE airflow_demo"'
          )
          
          tCreateTable = BashOperator(
              task_id='ch_table',
              depends_on_past=False,
              bash_command='bytehouse-cli --host {HOST} --port {PORT} --secure --token {API_KEY} -q "CREATE TABLE airflow_demo.demo_tb(id UInt64) ENGINE=CnchMergeTree() ORDER BY id"'
          )
          
          tImport = BashOperator(
              task_id='ch_import',
              depends_on_past=False,
              bash_command='bytehouse-cli --host {HOST} --port {PORT} --secure --token {API_KEY} -q "INSERT INTO airflow_demo.demo_tb values (1)"',
          )
          
          tSelect = BashOperator(
              task_id='ch_select',
              depends_on_past=False,
              bash_command='bytehouse-cli --host {HOST} --port {PORT} --secure --token {API_KEY} -q "SELECT * FROM airflow_demo.user"',
          )
          tCreateDatabase >> tCreateTable >> tImport >> tSelect
   

   from datetime import timedelta
      import pendulum
      from airflow import DAG
      from airflow.operators.bash import BashOperator
      default_args = {
          'owner': 'airflow',
          'depends_on_past': False,
          'email': ['airflow@example.com'],
          'email_on_failure': False,
          'email_on_retry': False,
          'retries': 1,
          'retry_delay': timedelta(minutes=1),
      }
      # 创建 DAG 实例
      with DAG(
          'test_bytehouse',
          default_args=default_args,
          description='A simple tutorial DAG',
          schedule=timedelta(days=1),  
          start_date=pendulum.today('UTC').add(days=-1),
          tags=['example'],
      ) as dag:
          # 定义 BashOperators
          tCreateDatabase = BashOperator(
              task_id='ch_database',
              depends_on_past=False,
              bash_command='bytehouse-cli --user {accountID_or_accountName}::{username} --password {password} --host {HOST} --port 19000 --secure -q "CREATE DATABASE airflow_demo"'
          )
          
          tCreateTable = BashOperator(
              task_id='ch_table',
              depends_on_past=False,
              bash_command='bytehouse-cli --user {accountID_or_accountName}::{username} --password {password} --host {HOST} --port 19000 --secure -q "CREATE TABLE airflow_demo.demo_tb(id UInt64) ENGINE=CnchMergeTree() ORDER BY id"'
          )
          
          tImport = BashOperator(
              task_id='ch_import',
              depends_on_past=False,
              bash_command='bytehouse-cli --user {accountID_or_accountName}::{username} --password {password} --host {HOST} --port 19000 --secure -q "INSERT INTO airflow_demo.demo_tb values (1)"',
          )
          
          tSelect = BashOperator(
              task_id='ch_select',
              depends_on_past=False,
              bash_command='bytehouse-cli --user {accountID_or_accountName}::{username} --password {password} --host {HOST} --port 19000 --secure -q "SELECT * FROM airflow_demo.user"',
          )
          tCreateDatabase >> tCreateTable >> tImport >> tSelect
   

   参数	使用 IAM 用户连接	使用数据库用户连接
   host	配置为 ByteHouse 的网络域名，其中 {TENANT_ID}、{REGION} 分别为火山引擎主账号的账号 ID 和 ByteHouse 的地域信息，您可以在 ByteHouse 控制台的 租户管理 > 基础信息 > 租户信息 中查看对应信息。详情请参见步骤二：配置网络信息。
   port	配置为：8123。
   username & Password	username：无需配置。 password：您可在 **** ByteHouse 控制台的 租户管理> 基础信息中获取的 API Key。详情请参见获取 API Key。	username 配置为 {accountID_or_accountName}::{username}[::{envID}]，详情请参见步骤三：获取 ByteHouse 连接串信息： accountID_or_accountName：指火山引擎用户账号 ID 或名称，可登录 ByteHouse 控制台，单击右上角个人中心查看并复制账号ID 或名称。 username：登录 ByteHouse 数据库账号用户名。可在 ByteHouse 控制台的权限与生态 > 权限管理 > 用户 查看数据库用户名。 envID：可选配置，数据库所在的环境名称。如果使用 default 环境，可不配置，如需使用其他环境，需指定环境名称。您可登录 **** ByteHouse 控制台，在 **租户管理 > 基础信息 > 当前环境 **中获取。 password：数据库账号的密码由管理员创建数据库账号时自定义配置，您可联系管理员获取密码。如果密码丢失或遗忘，可通联系管理员重置密码，详情请参考重置密码。
   secure & verify	secure：配置为 true verify：配置为 false

3. 在当前文件路径下运行 python test_bytehouse.py，以在 Airflow 中创建 DAG。
   刷新浏览器中的网页，您可以看到新创建的 DAG 名为 test_bytehouse显示在 DAG 列表中。
   

4. 在终端中运行以下的 Airflow 命令，以查看 DAG 列表和 test_Bytehouse DAG 中的测试子任务。你可以分别测试查询执行和数据导入任务。

   #prints the list of tasks in the "test_bytehouse" DAG
   airflow tasks list test_bytehouse
   #output
   ch_database
   ch_import
   ch_select
   ch_table
   
   #prints the hierarchy of tasks in the "test_bytehouse" DAG
   airflow tasks list test_bytehouse --tree
   #output
   <Task(BashOperator): ch_database>
       <Task(BashOperator): ch_table>
           <Task(BashOperator): ch_import>
               <Task(BashOperator): ch_select>
   

5. 运行 DAG 成功之后，您会在 DAG 中看到成功标志。
   
   在您的 ByteHouse 账户中检查查询历史记录页面和数据库模块，您可以看到数据被成功查询/加载。