Connexions AWS Marketplace et personnalisées Options générales

Types et options de connexion pour ETL dans AWS Glue pour Spark

Dans AWS Glue pour Spark, diverses méthodes et transformations PySpark et Scala spécifient le type de connexion à l'aide d'un paramètre connectionType. Ils spécifient des options de connexion à l'aide d'un paramètre connectionOptions ou options.

Le paramètre connectionType peut prendre les valeurs indiquées dans le tableau suivant. Les valeurs de paramètre associées connectionOptions (ou options) pour chaque type sont documentées dans les sections suivantes. Sauf indication contraire, les paramètres s'appliquent lorsque la connexion est utilisée comme source ou comme collecteur.

Pour obtenir un exemple de code qui illustre le paramétrage et l’utilisation des options de connexion, consultez la page d’accueil de chaque type de connexion.

`connectionType`	Se connecte à
dynamodb	Base de données Amazon DynamoDB
kinesis	Amazon Kinesis Data Streams
s3	Amazon S3
documentdb	Base de données Amazon DocumentDB (compatible avec MongoDB)
openSearch	Amazon OpenSearch Service.
redshift	Base de données Amazon Redshift
kafka	Kafka ou Amazon Managed Streaming for Apache Kafka
azurecosmos	Azure Cosmos pour NoSQL.
azuresql	Azure SQL.
bigquery	Google BigQuery.
mongodb	Base de données MongoDB, y compris MongoDB Atlas.
sqlserver	Base de données Microsoft SQL Server (consultez Connexions JDBC)
mysql	Base de données MySQL (consultez Connexions JDBC)
oracle	Base de données Oracle (consultez Connexions JDBC)
postgresql	Base de données PostgreSQL (consultez Connexions JDBC)
saphana	SAP HANA.
snowflake	Lac de données Snowflake
teradata	Teradata Vantage.
vertica	Vertica.
personnalisé.*	Magasins de données Spark, Athena ou JDBC (voir Valeurs connectionType AWS Marketplace et personnalisées
marketplace.*	Magasins de données Spark, Athena ou JDBC (voir Valeurs connectionType AWS Marketplace et personnalisées)

Valeurs connectionType AWS Marketplace et personnalisées

Tel est le cas des éléments suivants :

"connectionType": "marketplace.athena" : désigne une connexion à un magasin de données Amazon Athena. La connexion utilise un connecteur de AWS Marketplace.
"connectionType": "marketplace.spark" : désigne une connexion à un magasin de données Apache Spark. La connexion utilise un connecteur de AWS Marketplace.
"connectionType": "marketplace.jdbc" : désigne une connexion à un magasin de données JDBC. La connexion utilise un connecteur de AWS Marketplace.
"connectionType": "custom.athena" : désigne une connexion à un magasin de données Amazon Athena. La connexion utilise un connecteur personnalisé que vous chargez sur AWS Glue Studio.
"connectionType": "custom.spark" : désigne une connexion à un magasin de données Apache Spark. La connexion utilise un connecteur personnalisé que vous chargez sur AWS Glue Studio.
"connectionType": "custom.jdbc" : désigne une connexion à un magasin de données JDBC. La connexion utilise un connecteur personnalisé que vous chargez sur AWS Glue Studio.

Options de connexion pour le type custom.jdbc ou marketplace.jdbc

className — chaîne, obligatoire, nom de la classe du pilote.
connectionName — chaîne, obligatoire, nom de la connexion associée au connecteur.
url — chaîne, obligatoire, URL JDBC avec des espaces réservés (${}) qui sont utilisés pour créer la connexion à la source de données. L'espace réservé ${secretKey} est remplacé par le secret du même nom dans AWS Secrets Manager. Reportez-vous à la documentation du magasin de données pour plus d'informations sur la construction de l'URL.
secretId ou user/password — chaîne, obligatoire, utilisée pour récupérer les informations d'identification de l'URL.
dbTable ou query — chaîne, obligatoire, la table ou la requête SQL à partir de laquelle obtenir les données. Vous pouvez préciser dbTable ou query, mais pas les deux.
partitionColumn — chaîne, facultatif, nom d'une colonne entière utilisée pour le partitionnement. Cette option fonctionne uniquement lorsqu'elle est incluse dans lowerBound, upperBound et numPartitions. Cette option fonctionne de la même manière que dans le lecteur JDBC SQL Spark. Pour de plus amples informations, veuillez consulter JJDBC To Other Databasesdans le document Apache Spark SQL, DataFrames and Datasets Guide.

Les valeurs lowerBound et upperBound sont utilisées pour décider de la progression de la partition, pas pour filtrer les lignes de la table. Toutes les lignes de la table sont partitionnées et renvoyées.
Note
Lorsque vous utilisez une requête au lieu d'un nom de table, vous devez valider que la requête fonctionne avec la condition de partitionnement spécifiée. Par exemple :
- Si le format de votre requête est "SELECT col1 FROM table1", testez la requête en ajoutant une clause WHERE à la fin de la requête qui utilise la colonne de partition.
- Si le format de votre requête est SELECT col1 FROM table1 WHERE col2=val", testez la requête en étendant la clause WHERE avec AND et une expression qui utilise la colonne de partition.
lowerBound — entier, facultatif, valeur minimale de partitionColumn qui est utilisée pour décider de la progression de la partition.
upperBound — entier, facultatif, valeur maximale de partitionColumn qui est utilisée pour décider de la progression de la partition.
numPartitions — entier, facultatif, nombre de partitions. Cette valeur, ainsi que lowerBound (inclusive) et upperBound (exclusive) forment les progressions de partition pour les expressions de clause WHERE générées qui sont utilisées pour diviser le fichier partitionColumn.

Important
Soyez prudent avec le nombre de partitions, car avoir un trop grand nombre de partitions peut causer des problèmes sur vos systèmes de bases de données externes.
filterPredicate — chaîne, facultative, clause de condition supplémentaire pour filtrer les données à partir de la source. Par exemple :
```
BillingCity='Mountain View'
```
Lorsque vous utilisez une requête au lieu d'un nom de table, vous devez vérifier que la requête fonctionne avec le filterPredicate spécifié. Par exemple :
- Si le format de votre requête est "SELECT col1 FROM table1", testez la requête en ajoutant une clause WHERE à la fin de la requête qui utilise le prédicat de filtre.
- Si le format de votre requête est "SELECT col1 FROM table1 WHERE col2=val", testez la requête en étendant la clause WHERE avec AND et une expression qui utilise le prédicat de filtre.
dataTypeMapping — dictionnaire, facultatif, mappage de type de données personnalisé qui crée un mappage d'un type de données JDBC à un type de données Glue. Par exemple, l'option "dataTypeMapping":{"FLOAT":"STRING"} mappe les champs de données du type JDBC FLOAT sur le type Java String en appelant la méthode ResultSet.getString() du pilote, et l'utilise pour créer les enregistrements AWS Glue. L'objet est ResultSet implémenté par chaque pilote, donc le comportement est spécifique au pilote que vous utilisez. Reportez-vous à la documentation de votre pilote JDBC pour comprendre comment le pilote effectue les conversions.
Les types de données AWS Glue actuellement pris en charge sont les suivants :
- DATE
- CHAÎNE
- TIMESTAMP
- INT
- FLOAT
- LONG
- BIGDECIMAL
- BYTE
- SHORT
- DOUBLE
Les types de données JDBC pris en charge sont Java8 java.sql.types.

Les mappages de type de données par défaut (de JDBC à AWS Glue) sont les suivants :
- DATE -> DATE
- VARCHAR -> STRING
- CHAR -> STRING
- LONGNVARCHAR -> STRING
- TIMESTAMP -> TIMESTAMP
- INTEGER -> INT
- FLOAT -> FLOAT
- REAL -> FLOAT
- BIT -> BOOLEAN
- BOOLEAN -> BOOLEAN
- BIGINT -> LONG
- DECIMAL -> BIGDECIMAL
- NUMERIC -> BIGDECIMAL
- TINYINT -> SHORT
- SMALLINT -> SHORT
- DOUBLE -> DOUBLE
Si vous utilisez un mappage de type de données personnalisé avec l'option dataTypeMapping, vous pouvez remplacer un mappage de type de données par défaut. Seuls les types de données JDBC répertoriés dans l'option dataTypeMapping sont affectés ; le mappage par défaut est utilisé pour tous les autres types de données JDBC. Vous pouvez ajouter des mappages pour des types de données JDBC supplémentaires si nécessaire. Si un type de données JDBC n'est inclus ni dans le mappage par défaut ni dans un mappage personnalisé, le type de données est converti en type de données AWS Glue STRING par défaut.

Les exemples de code Python suivants montrent comment lire des bases de données JDBC avec des pilotes JDBC AWS Marketplace. Il montre la lecture à partir d'une base de données et l'écriture dans un emplacement S3.


    import sys
    from awsglue.transforms import *
    from awsglue.utils import getResolvedOptions
    from pyspark.context import SparkContext
    from awsglue.context import GlueContext
    from awsglue.job import Job
     
    ## @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)
    ## @type: DataSource
    ## @args: [connection_type = "marketplace.jdbc", connection_options = 
     {"dataTypeMapping":{"INTEGER":"STRING"},"upperBound":"200","query":"select id, 
       name, department from department where id < 200","numPartitions":"4",
       "partitionColumn":"id","lowerBound":"0","connectionName":"test-connection-jdbc"},
        transformation_ctx = "DataSource0"]
    ## @return: DataSource0
    ## @inputs: []
    DataSource0 = glueContext.create_dynamic_frame.from_options(connection_type = 
      "marketplace.jdbc", connection_options = {"dataTypeMapping":{"INTEGER":"STRING"},
      "upperBound":"200","query":"select id, name, department from department where 
       id < 200","numPartitions":"4","partitionColumn":"id","lowerBound":"0",
       "connectionName":"test-connection-jdbc"}, transformation_ctx = "DataSource0")
    ## @type: ApplyMapping
    ## @args: [mappings = [("department", "string", "department", "string"), ("name", "string",
      "name", "string"), ("id", "int", "id", "int")], transformation_ctx = "Transform0"]
    ## @return: Transform0
    ## @inputs: [frame = DataSource0]
    Transform0 = ApplyMapping.apply(frame = DataSource0, mappings = [("department", "string",
      "department", "string"), ("name", "string", "name", "string"), ("id", "int", "id", "int")], 
       transformation_ctx = "Transform0")
    ## @type: DataSink
    ## @args: [connection_type = "s3", format = "json", connection_options = {"path": 
     "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0"]
    ## @return: DataSink0
    ## @inputs: [frame = Transform0]
    DataSink0 = glueContext.write_dynamic_frame.from_options(frame = Transform0, 
      connection_type = "s3", format = "json", connection_options = {"path": 
      "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0")
    job.commit()

Options de connexion pour le type custom.athena ou marketplace.athena

className — chaîne, obligatoire, nom de la classe du pilote. Lorsque vous utilisez le connecteur Athena-CloudWatch, cette valeur de paramètre est le préfixe du nom de classe (par exemple, "com.amazonaws.athena.connectors"). Le connecteur Athena-CloudWatch est composé de deux classes : un gestionnaire de métadonnées et un gestionnaire d'enregistrements. Si vous fournissez le préfixe commun ici, l'API charge les classes correctes en fonction de ce préfixe.
tableName — chaîne, obligatoire, nom du flux de journal CloudWatch à lire. Cet extrait de code utilise le nom de vue spécial all_log_streams, ce qui signifie que la trame de données dynamique renvoyée contiendra les données de tous les flux de journaux du groupe de journaux.
schemaName — chaîne, obligatoire, nom du groupe de journaux CloudWatch à partir duquel lire. Par exemple, /aws-glue/jobs/output.
connectionName — chaîne, obligatoire, nom de la connexion associée au connecteur.

Pour des options supplémentaires pour ce connecteur, consultez le fichier Amazon Athena CloudWatch Connector README sur GitHub.

L'exemple de code Python suivant montre comment lire à partir d'un magasin de données Athena à l'aide d'un connecteur AWS Marketplace. Il montre la lecture à partir d'Athena et l'écriture dans un emplacement S3.


    import sys
    from awsglue.transforms import *
    from awsglue.utils import getResolvedOptions
    from pyspark.context import SparkContext
    from awsglue.context import GlueContext
    from awsglue.job import Job
     
    ## @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)
    ## @type: DataSource
    ## @args: [connection_type = "marketplace.athena", connection_options = 
     {"tableName":"all_log_streams","schemaName":"/aws-glue/jobs/output",
      "connectionName":"test-connection-athena"}, transformation_ctx = "DataSource0"]
    ## @return: DataSource0
    ## @inputs: []
    DataSource0 = glueContext.create_dynamic_frame.from_options(connection_type = 
      "marketplace.athena", connection_options = {"tableName":"all_log_streams",,
      "schemaName":"/aws-glue/jobs/output","connectionName":
      "test-connection-athena"}, transformation_ctx = "DataSource0")
    ## @type: ApplyMapping
    ## @args: [mappings = [("department", "string", "department", "string"), ("name", "string",
      "name", "string"), ("id", "int", "id", "int")], transformation_ctx = "Transform0"]
    ## @return: Transform0
    ## @inputs: [frame = DataSource0]
    Transform0 = ApplyMapping.apply(frame = DataSource0, mappings = [("department", "string",
      "department", "string"), ("name", "string", "name", "string"), ("id", "int", "id", "int")], 
       transformation_ctx = "Transform0")
    ## @type: DataSink
    ## @args: [connection_type = "s3", format = "json", connection_options = {"path": 
     "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0"]
    ## @return: DataSink0
    ## @inputs: [frame = Transform0]
    DataSink0 = glueContext.write_dynamic_frame.from_options(frame = Transform0, 
      connection_type = "s3", format = "json", connection_options = {"path": 
      "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0")
    job.commit()

Options de connexion pour le type custom.spark ou marketplace.spark

className — chaîne, obligatoire, nom de la classe du connecteur.
secretId — chaîne, facultative, utilisée pour récupérer les informations d'identification pour la connexion du connecteur.
connectionName — chaîne, obligatoire, nom de la connexion associée au connecteur.
D'autres options dépendent du magasin de données. Par exemple, les options de configuration OpenSearch commencent par le préfixe es, comme décrit dans la documentation Elasticsearch for Apache Hadoop. Les connexions Spark à Snowflake utilisent des options telles que sfUser et sfPassword, comme décrit dans Using the Spark Connector dans le guide Connecting to Snowflake.

L'exemple de code Python suivant montre comment lire à partir d'un magasin de données OpenSearch à l'aide d'une connexion marketplace.spark.


    import sys
    from awsglue.transforms import *
    from awsglue.utils import getResolvedOptions
    from pyspark.context import SparkContext
    from awsglue.context import GlueContext
    from awsglue.job import Job
     
    ## @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)
    ## @type: DataSource
    ## @args: [connection_type = "marketplace.spark", connection_options = {"path":"test",
      "es.nodes.wan.only":"true","es.nodes":"https://<AWS endpoint>",
      "connectionName":"test-spark-es","es.port":"443"}, transformation_ctx = "DataSource0"]
    ## @return: DataSource0
    ## @inputs: []
    DataSource0 = glueContext.create_dynamic_frame.from_options(connection_type = 
      "marketplace.spark", connection_options = {"path":"test","es.nodes.wan.only":
      "true","es.nodes":"https://<AWS endpoint>","connectionName":
      "test-spark-es","es.port":"443"}, transformation_ctx = "DataSource0")
    ## @type: DataSink
    ## @args: [connection_type = "s3", format = "json", connection_options = {"path": 
         "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0"]
    ## @return: DataSink0
    ## @inputs: [frame = DataSource0]
    DataSink0 = glueContext.write_dynamic_frame.from_options(frame = DataSource0, 
       connection_type = "s3", format = "json", connection_options = {"path": 
       "s3://<S3 path>/", "partitionKeys": []}, transformation_ctx = "DataSink0")
    job.commit()

Options générales

Les options de cette section sont fournies en tant que connection_options, mais ne s'appliquent pas à un connecteur en particulier.

Les paramètres suivants sont généralement utilisés lors de la configuration de signets. Ils peuvent s'appliquer aux flux de travail Amazon S3 ou JDBC. Pour de plus amples informations, veuillez consulter Utilisation des marque-pages de tâche.

jobBookmarkKeys : un tableau des noms des colonnes.
jobBookmarkKeysSortOrder : chaîne définissant comment comparer les valeurs en fonction de l'ordre de tri. Valeurs valides : "asc", "desc".
useS3ListImplementation : utilisé pour gérer les performances de la mémoire lors de la liste du contenu du compartiment Amazon S3. Pour plus d'informations, consultez Optimize memory management in AWS Glue.

Avertissement JavaScript est désactivé ou n'est pas disponible dans votre navigateur.

Pour que vous puissiez utiliser la documentation AWS, Javascript doit être activé. Vous trouverez des instructions sur les pages d'aide de votre navigateur.

Conventions de rédaction

Fonctionnalités et optimisations

Connexions DynamoDB