Connessioni Kinesis - AWS Glue
Configurazione di KinesisLettura da KinesisScrittura su KinesisParametri di connessione Kinesis

Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Connessioni Kinesis

È possibile utilizzare una connessione Kinesis per leggere e scrivere su flussi di dati Amazon Kinesis utilizzando le informazioni memorizzate in una tabella Data Catalog o fornendo informazioni per accedere direttamente al flusso di dati. Puoi leggere le informazioni da Kinesis in Spark DataFrame, quindi convertirle in un Glue. AWS DynamicFrame È possibile DynamicFrames scrivere su Kinesis in un JSON formato. Se accedi direttamente al flusso di dati, utilizza queste opzioni per fornire le informazioni su come accedere al flusso di dati.

Se utilizzi getCatalogSource o create_data_frame_from_catalog per consumare i registri da una sorgente di streaming Kinesis, il processo avrà le informazioni sul database catalogo dati e sul nome della tabella, e potrà usarle per ottenere alcuni parametri di base per la lettura dalla sorgente di streaming Kinesis. Se utilizzi getSource, getSourceWithFormat, createDataFrameFromOptions o create_data_frame_from_options, dovrai specificare questi parametri di base utilizzando le opzioni di connessione descritte qui.

È possibile specificare le opzioni di connessione per Kinesis utilizzando i seguenti argomenti per i metodi specificati nella classe GlueContext.

  • Scala

    • connectionOptions: utilizza con getSource, createDataFrameFromOptions e getSink

    • additionalOptions: utilizza con getCatalogSource, getCatalogSink

    • options: utilizza con getSourceWithFormat, getSinkWithFormat

  • Python

    • connection_options: utilizza con create_data_frame_from_options, write_dynamic_frame_from_options

    • additional_options: utilizza con create_data_frame_from_catalog, write_dynamic_frame_from_catalog

    • options: utilizza con getSource, getSink

Per note e restrizioni sui ETL lavori di Streaming, consultaStreaming di note e restrizioni ETL.

Configurazione di Kinesis

Per connetterti a un flusso di dati Kinesis in un job AWS Glue Spark, avrai bisogno di alcuni prerequisiti:

  • Se è in lettura, il job AWS Glue deve disporre IAM delle autorizzazioni di accesso Read per il flusso di dati Kinesis.

  • In fase di scrittura, il job AWS Glue deve disporre IAM delle autorizzazioni di livello di accesso Write per il flusso di dati Kinesis.

In alcuni casi, è necessario configurare ulteriori prerequisiti:

  • Se il tuo job AWS Glue è configurato con connessioni di rete aggiuntive (in genere per connettersi ad altri set di dati) e una di queste connessioni offre opzioni di VPC rete Amazon, questo indirizzerà il tuo job alla comunicazione su AmazonVPC. In questo caso dovrai anche configurare il flusso di dati Kinesis per comunicare su Amazon. VPC Puoi farlo creando un VPC endpoint di interfaccia tra il tuo flusso di dati Amazon VPC e Kinesis. Per ulteriori informazioni, consulta Using Kinesis Data Streams VPC with Interface Endpoints.

  • Quando si specifica un flusso di dati Amazon Kinesis in un altro account, è necessario impostare i ruoli e le policy per consentire l'accesso multi-account. Per ulteriori informazioni, consulta Esempio: lettura da un flusso Kinesis in un account diverso.

Per ulteriori informazioni sui prerequisiti del ETL lavoro in Streaming, consulta. Aggiunta di processi di streaming ETL in AWS Glue

Esempio: lettura da flussi Kinesis

Esempio: lettura da flussi Kinesis

Usato in combinazione con forEachBatch.

Esempio per l'origine di streaming Amazon Kinesis:

kinesis_options =
   { "streamARN": "arn:aws:kinesis:us-east-2:777788889999:stream/fromOptionsStream",
     "startingPosition": "TRIM_HORIZON", 
     "inferSchema": "true", 
     "classification": "json" 
   }
data_frame_datasource0 = glueContext.create_data_frame.from_options(connection_type="kinesis", connection_options=kinesis_options)

Esempio: scrittura su flussi Kinesis

Esempio: lettura da flussi Kinesis

Usato in combinazione con forEachBatch.

Esempio per l'origine di streaming Amazon Kinesis:

kinesis_options =
   { "streamARN": "arn:aws:kinesis:us-east-2:777788889999:stream/fromOptionsStream",
     "startingPosition": "TRIM_HORIZON", 
     "inferSchema": "true", 
     "classification": "json" 
   }
data_frame_datasource0 = glueContext.create_data_frame.from_options(connection_type="kinesis", connection_options=kinesis_options)

Indicazioni di riferimento alle opzioni di connessione a Kinesis

Indica le opzioni di connessione ad Amazon Kinesis Data Streams.

Utilizza le seguenti opzioni di connessione per le origini dati in streaming Kinesis:

  • "streamARN": (obbligatorio) utilizzato per la lettura/scrittura. Il ARN flusso di dati Kinesis.

  • "classification": (obbligatorio per la lettura) utilizzato per la lettura. Il formato di file utilizzato dai dati nel record. Obbligatorio, a meno che non sia fornito tramite Catalogo dati.

  • "streamName": (facoltativo) utilizzato per la lettura. Il nome di un flusso di dati Kinesis da cui leggere. Usato con endpointUrl.

  • "endpointUrl": (facoltativo) utilizzato per la lettura. Predefinito: "https://kinesis.us-east-1.amazonaws.com». L' AWS endpoint del flusso Kinesis. Non è necessario modificarlo a meno che non ci si stia connettendo a una regione speciale.

  • "partitionKey": (facoltativo) utilizzato per la scrittura. La chiave di partizione di Kinesis utilizzata per la produzione dei record.

  • "delimiter": (facoltativo) utilizzato per la lettura. Il separatore di valori usato quando è. classification CSV Il valore predefinito è ",".

  • "startingPosition": (facoltativo) utilizzato per la lettura. La posizione di partenza nel flusso dei dati Kinesis da cui leggere i dati. I valori possibili sono"latest", "trim_horizon""earliest", o una stringa Timestamp nel UTC formato nel modello yyyy-mm-ddTHH:MM:SSZ (dove Z rappresenta un offset di UTC fuso orario con un +/-. Ad esempio «2023-04-04T 08:00:00-04:00 «). Il valore predefinito è "latest". Nota: la stringa Timestamp in UTC Format for "startingPosition" è supportata solo per AWS Glue versione 4.0 o successiva.

  • "failOnDataLoss": (facoltativo) non è possibile eseguire il processo se una partizione attiva è mancante o scaduta. Il valore predefinito è "false".

  • "awsSTSRoleARN": (facoltativo) utilizzato per la lettura/scrittura. L'Amazon Resource Name (ARN) del ruolo da assumere utilizzando AWS Security Token Service (AWS STS). Questo ruolo deve disporre delle autorizzazioni per descrivere o leggere le operazioni dei registri per il flusso di dati Kinesis. Quando si accede a un flusso di dati in un altro account, è necessario utilizzare questo parametro. Usato in combinazione con "awsSTSSessionName".

  • "awsSTSSessionName": (facoltativo) utilizzato per la lettura/scrittura. Un identificatore della sessione che assume il ruolo usando AWS STS. Quando si accede a un flusso di dati in un altro account, è necessario utilizzare questo parametro. Usato in combinazione con "awsSTSRoleARN".

  • "awsSTSEndpoint": (Facoltativo) L' AWS STS endpoint da utilizzare quando ci si connette a Kinesis con un ruolo presunto. Ciò consente di utilizzare l' AWS STS endpoint regionale in aVPC, cosa non possibile con l'endpoint globale predefinito.

  • "maxFetchTimeInMs": (facoltativo) utilizzato per la lettura. Il tempo massimo impiegato dall'esecutore del lavoro per leggere i record del batch corrente dal flusso di dati Kinesis, specificato in millisecondi (ms). È possibile GetRecords API effettuare più chiamate entro questo periodo. Il valore predefinito è 1000.

  • "maxFetchRecordsPerShard": (facoltativo) utilizzato per la lettura. Il numero massimo di record da recuperare per shard nel flusso di dati Kinesis per microbatch. Nota: il client può superare questo limite se il job di streaming ha già letto record aggiuntivi da Kinesis (nella stessa chiamata get-records). Se maxFetchRecordsPerShard deve essere rigoroso, deve essere un multiplo di. maxRecordPerRead Il valore predefinito è 100000.

  • "maxRecordPerRead": (facoltativo) utilizzato per la lettura. Il numero massimo di record da recuperare nel flusso di dati Kinesis in ciascuna operazione getRecords. Il valore predefinito è 10000.

  • "addIdleTimeBetweenReads": (facoltativo) utilizzato per la lettura. Aggiunge un ritardo tra due operazioni consecutive getRecords. Il valore predefinito è "False". Questa opzione è configurabile solo per Glue versione 2.0 e successive.

  • "idleTimeBetweenReadsInMs": (facoltativo) utilizzato per la lettura. Il ritardo minimo tra due operazioni consecutive getRecords, specificato in ms. Il valore predefinito è 1000. Questa opzione è configurabile solo per Glue versione 2.0 e successive.

  • "describeShardInterval": (facoltativo) utilizzato per la lettura. L'intervallo di tempo minimo tra due ListShards API chiamate entro il quale lo script deve prendere in considerazione il resharding. Per ulteriori informazioni, consulta Strategie per il resharding nella Guida per gli sviluppatori di Amazon Kinesis Data Streams. Il valore predefinito è 1s.

  • "numRetries": (facoltativo) utilizzato per la lettura. Il numero massimo di tentativi per le richieste Kinesis Data Streams. API Il valore predefinito è 3.

  • "retryIntervalMs": (facoltativo) utilizzato per la lettura. Il periodo di riflessione (specificato in ms) prima di riprovare la chiamata Kinesis Data Streams. API Il valore predefinito è 1000.

  • "maxRetryIntervalMs": (facoltativo) utilizzato per la lettura. Il periodo di raffreddamento massimo (specificato in ms) tra due tentativi di una chiamata Kinesis Data Streams. API Il valore predefinito è 10000.

  • "avoidEmptyBatches": (facoltativo) utilizzato per la lettura. Impedisce la creazione di un processo microbatch vuoto controllando la presenza di dati non letti nel flusso dei dati Kinesis prima che il batch venga avviato. Il valore predefinito è "False".

  • "schema": (Obbligatorio se inferSchema impostato su false) Utilizzato per la lettura. Lo schema da utilizzare per elaborare il payload. Se la classificazione è avro, lo schema fornito dovrà essere nel formato dello schema Avro. Se la classificazione non è validaavro, lo schema fornito deve essere nel formato dello DDL schema.

    Di seguito sono riportati alcuni esempi di schema.

    Example in DDL schema format
    `column1` INT, `column2` STRING , `column3` FLOAT
    Example in Avro schema format
    {
  "type":"array",
  "items":
  {
    "type":"record",
    "name":"test",
    "fields":
    [
      {
        "name":"_id",
        "type":"string"
      },
      {
        "name":"index",
        "type":
        [
          "int",
          "string",
          "float"
        ]
      }
    ]
  }
}

  • "inferSchema": (facoltativo) utilizzato per la lettura. Il valore predefinito è "false". Se impostato su "true", lo schema verrà rilevato in fase di runtime dal payload all'interno di foreachbatch.

  • "avroSchema": (obsoleto) utilizzato per la lettura. Parametro utilizzato per specificare uno schema di dati Avro quando viene utilizzato il formato Avro. Questo parametro è obsoleto. Utilizzo del parametro schema.

  • "addRecordTimestamp": (facoltativo) utilizzato per la lettura. Quando questa opzione è impostata su "true", l'output dei dati conterrà una colonna aggiuntiva denominata "__src_timestamp" che indica l'ora in cui il record corrispondente è stato ricevuto dal flusso. Il valore predefinito è "false". Questa opzione è supportata in AWS Glue versione 4.0 o successive.

  • "emitConsumerLagMetrics": (facoltativo) utilizzato per la lettura. Quando l'opzione è impostata su «true», per ogni batch emette le metriche relative alla durata compresa tra il record più vecchio ricevuto dallo stream e il momento in AWS Glue cui arriva. CloudWatch Il nome della metrica è «glue.driver.streaming. maxConsumerLagInMs». Il valore predefinito è "false". Questa opzione è supportata in AWS Glue versione 4.0 o successive.

  • "fanoutConsumerARN": (facoltativo) utilizzato per la lettura. Il ARN nome di un utente di stream Kinesis per lo stream specificato in. streamARN Utilizzato per abilitare la modalità di fan-out avanzato per la connessione Kinesis. Per ulteriori informazioni sull'utilizzo di un flusso Kinesis con fan-out avanzato, consulta la pagina Utilizzo del fan-out avanzato nei processi di flussi di dati Kinesis.

  • "recordMaxBufferedTime": (facoltativo) utilizzato per la scrittura. Predefinito: 1000 (ms). Tempo massimo di memorizzazione nel buffer di un record in attesa di essere scritto.

  • "aggregationEnabled": (facoltativo) utilizzato per la scrittura. Default: true (VERO). Speciifica se i record devono essere aggregati prima di inviarli a Kinesis.

  • "aggregationMaxSize": (facoltativo) utilizzato per la scrittura. Impostazione predefinita: 51200 (byte). Se un record è superiore a questo limite, ignorerà l'aggregatore. Ricorda che Kinesis impone un limite di 50 KB alla dimensione del record. Se imposti questo valore oltre i 50 KB, i record di grandi dimensioni verranno rifiutati da Kinesis.

  • "aggregationMaxCount": (facoltativo) utilizzato per la scrittura. Predefinito: 4294967295. Numero massimo di voci da inserire in un record aggregato.

  • "producerRateLimit": (facoltativo) utilizzato per la scrittura. Predefinito: 150 (%). Limita la velocità di trasmissione effettiva per partizione inviata da un singolo produttore (ad esempio, il tuo processo), come percentuale del limite di backend.

  • "collectionMaxCount": (facoltativo) utilizzato per la scrittura. Predefinito: 500. Numero massimo di articoli da imballare in una PutRecords richiesta.

  • "collectionMaxSize": (facoltativo) utilizzato per la scrittura. Impostazione predefinita: 5242880 (byte). Quantità massima di dati da inviare con una PutRecords richiesta.