Utilisation d'Amazon MWAA avec Amazon RDS pour Microsoft SQL Server

Vous pouvez utiliser Amazon Managed Workflows for Apache Airflow pour vous connecter à un SQLserveur RDS for. L'exemple de code suivant est utilisé DAGs dans un environnement Amazon Managed Workflows pour Apache Airflow pour se connecter à un SQL serveur Amazon RDS pour Microsoft et y exécuter des requêtes.

Version

• L'exemple de code présenté sur cette page peut être utilisé avec Apache Airflow v1 en Python 3.7.

• Vous pouvez utiliser l'exemple de code présenté sur cette page avec Apache Airflow v2 en Python 3.10.

Prérequis

Pour utiliser l'exemple de code présenté sur cette page, vous aurez besoin des éléments suivants :

• Un MWAAenvironnement Amazon.

• Amazon MWAA et le SQL serveur RDS for s'exécutent sur le même AmazonVPC/

• VPCles groupes de sécurité d'Amazon MWAA et du serveur sont configurés avec les connexions suivantes :

  • Une règle entrante pour le port 1433 ouvert pour Amazon RDS dans le groupe MWAA de sécurité d'Amazon

  • Ou une règle de sortie pour le port d'1433ouverture d'Amazon MWAA vers RDS

• Apache Airflow Connection for RDS for SQL Server reflète le nom d'hôte, le port, le nom d'utilisateur et le mot de passe de la base de données RDS SQL du serveur Amazon créée lors du processus précédent.

Dépendances

Pour utiliser l'exemple de code présenté dans cette section, ajoutez la dépendance suivante à votrerequirements.txt. Pour en savoir plus, consultez Installation des dépendances Python

Apache Airflow v2

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

Apache Airflow v1

apache-airflow[mssql]==1.10.12

Connexion Apache Airflow v2

Si vous utilisez une connexion dans Apache Airflow v2, assurez-vous que l'objet de connexion Airflow inclut les paires clé-valeur suivantes :

1. Identifiant de connexion : mssql_default

2. Type de connexion : Amazon Web Services

3. Hôte : YOUR_DB_HOST

4. Schéma :

5. Identifiant : admin

6. Mot de passe :

7. Hafen : 1433

8. Supplémentaire :

Exemple de code

1. Dans votre invite de commande, accédez au répertoire dans lequel votre DAG code est stocké. Par exemple :

   cd dags

2. Copiez le contenu de l'exemple de code suivant et enregistrez-le localement soussql-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

Quelle est la prochaine étape ?

• Découvrez comment charger le requirements.txt fichier dans cet exemple dans votre compartiment Amazon S3 dansInstallation des dépendances Python.

• Découvrez comment télécharger le DAG code de cet exemple dans le dags dossier de votre compartiment Amazon S3 dansAjout ou mise à jour des DAG.

• Explorez des exemples de scripts et d'autres exemples de modules pymssql.

• Pour en savoir plus sur l'exécution de SQL code dans une SQL base de données Microsoft spécifique à l'aide du mssql_operator, consultez le guide de référence Apache Airflow.