本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。

# 将 Amazon RDS for Microsoft SQL Server 与 Amazon MWAA 一起使用
<a name="samples-sql-server"></a>

您可以使用 Amazon MWAA 连接到 [RDS for SQL Server](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/CHAP_SQLServer.html)。以下示例代码使用 Amazon Managed Workflows for Apache Airflow 环境中的 DAG 连接到 Amazon RDS for Microsoft SQL Server 并在其上执行查询。

**Topics**
+ [版本](#samples-sql-server-version)
+ [先决条件](#samples-sql-server-prereqs)
+ [依赖项](#samples-sql-server-dependencies)
+ [Apache Airflow v2 连接](#samples-sql-server-conn)
+ [代码示例](#samples-sql-server-code)
+ [接下来做什么？](#samples-sql-server-next-up)

## 版本
<a name="samples-sql-server-version"></a>

您可以在 [Python 3.10](https://peps.python.org/pep-0619/) 中将本页上的代码示例与 **Apache Airflow v2** 一起使用，在 [Python 3.11](https://peps.python.org/pep-0664/) 中与 **Apache Airflow v3** 一起使用。

## 先决条件
<a name="samples-sql-server-prereqs"></a>

要使用本页上的示例代码，您需要以下内容：
+ [Amazon MWAA 环境。](get-started.md)
+ Amazon MWAA 和 RDS for SQL Server 在同一个 Amazon VPC/上运行
+ Amazon MWAA 和服务器的 VPC 安全组配置有以下连接：
  + Amazon MWAA 安全组中对 Amazon RDS `1433` 开放端口的入站规则
  + 或者是从 Amazon MWAA 到 RDS 开放端口 `1433` 的出站规则
+ RDS for SQL Server 的 Apache Airflow 连接反映了在之前过程中创建的 Amazon RDS SQL 服务器数据库的主机名、端口、用户名和密码。

## 依赖项
<a name="samples-sql-server-dependencies"></a>

要使用本节中的示例代码，请将以下依赖项添加到 `requirements.txt`。要了解更多信息，请参阅 [安装 Python 依赖项](working-dags-dependencies.md)。

```
apache-airflow-providers-microsoft-mssql==1.0.1
			apache-airflow-providers-odbc==1.0.1
			pymssql==2.2.1
```

## Apache Airflow v2 连接
<a name="samples-sql-server-conn"></a>

如果您在 Apache Airflow v2 中使用连接，请确保 Airflow 连接对象包含以下键值对：

1. **连接 ID：**mssql\$1default

1. **连接类型：**Amazon Web Services

1. **主机：**`YOUR_DB_HOST`

1. **架构：**

1. **登录：**管理员

1. **密码：**

1. **端口：**1433

1. **附加依赖项：**

## 代码示例
<a name="samples-sql-server-code"></a>

1. 在命令提示符下，导航到存储 DAG 代码的目录。例如：

   ```
   cd dags
   ```

1. 复制以下代码示例的内容，并在本地另存为 `sql-server.py`。

   ```
   """
   Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
   Permission is hereby granted, free of charge, to any person obtaining a copy of
   this software and associated documentation files (the "Software"), to deal in
   the Software without restriction, including without limitation the rights to
   use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
   the Software, and to permit persons to whom the Software is furnished to do so.
   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
   FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
   COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
   IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
   CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
   """
   import pymssql
   import logging
   import sys
   from airflow import DAG
   from datetime import datetime
   from airflow.operators.mssql_operator import MsSqlOperator
   from airflow.operators.python_operator import PythonOperator
   
   default_args = {
       'owner': 'aws',
       'depends_on_past': False,
       'start_date': datetime(2019, 2, 20),
       'provide_context': True
   }
   
   dag = DAG(
       'mssql_conn_example', default_args=default_args, schedule_interval=None)
       
   drop_db = MsSqlOperator(
      task_id="drop_db",
      sql="DROP DATABASE IF EXISTS testdb;",
      mssql_conn_id="mssql_default",
      autocommit=True,
      dag=dag
   )
   
   create_db = MsSqlOperator(
      task_id="create_db",
      sql="create database testdb;",
      mssql_conn_id="mssql_default",
      autocommit=True,
      dag=dag
   )
   
   create_table = MsSqlOperator(
      task_id="create_table",
      sql="CREATE TABLE testdb.dbo.pet (name VARCHAR(20), owner VARCHAR(20));",
      mssql_conn_id="mssql_default",
      autocommit=True,
      dag=dag
   )
   
   insert_into_table = MsSqlOperator(
      task_id="insert_into_table",
      sql="INSERT INTO testdb.dbo.pet VALUES ('Olaf', 'Disney');",
      mssql_conn_id="mssql_default",
      autocommit=True,
      dag=dag
   )
   
   def select_pet(**kwargs):
      try:
           conn = pymssql.connect(
               server='sampledb.<xxxxxx>.<region>.rds.amazonaws.com',
               user='admin',
               password='<yoursupersecretpassword>',
               database='testdb'
           )
           
           # Create a cursor from the connection
           cursor = conn.cursor()
           cursor.execute("SELECT * from testdb.dbo.pet")
           row = cursor.fetchone()
           
           if row:
               print(row)
      except:
         logging.error("Error when creating pymssql database connection: %s", sys.exc_info()[0])
   
   select_query = PythonOperator(
       task_id='select_query',
       python_callable=select_pet,
       dag=dag,
   )
   
   drop_db >> create_db >> create_table >> insert_into_table >> select_query
   ```

## 接下来做什么？
<a name="samples-sql-server-next-up"></a>
+ 要了解如何将本示例中的 `requirements.txt` 文件上传到 Amazon S3 存储桶，请参阅 [安装 Python 依赖项](working-dags-dependencies.md)。
+ 要了解如何将本示例中的 DAG 代码上传到 Amazon S3 存储桶的 `dags` 文件夹，请参阅 [添加或更新 DAG](configuring-dag-folder.md)。
+ 浏览示例脚本和其他 [pymssql 模块示例](https://pymssql.readthedocs.io/en/stable/pymssql_examples.html)。
+ 在*《Apache Airflow 参考指南》*中详细了解如何使用 [mssql\$1operator](https://airflow.apache.org/docs/apache-airflow/1.10.12/_api/airflow/operators/mssql_operator/index.html?highlight=mssqloperator#airflow.operators.mssql_operator.MsSqlOperator) 在特定的 Microsoft SQL 数据库中执行 SQL 代码。