Amazon Kinesis - Amazon Timestream
Verwenden Amazon Managed Service für Apache FlinkPipes verwenden EventBridge

Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Amazon Kinesis

Sie können Daten von Kinesis Data Streams an Timestream senden, um den Timestream Beispieldatenconnector für Managed Service für Apache Flink zu LiveAnalytics verwenden. Weitere Informationen finden Sie Amazon Managed Service für Apache Flink unter Apache Flink.

Verwenden von EventBridge Pipes zum Senden von Kinesis-Daten an Timestream

Sie können EventBridge Pipes verwenden, um Daten von einem Kinesis-Stream an einen Amazon Timestream für LiveAnalytics eine Tabelle zu senden.

Pipes sind für point-to-point Integrationen zwischen unterstützten Quellen und Zielen vorgesehen und unterstützen erweiterte Transformationen und Anreicherungen. Pipes reduzieren den Bedarf an Fachwissen und Integrationscode bei der Entwicklung ereignisgesteuerter Architekturen. Zum Einrichten einer Pipe wählen Sie die Quelle aus, fügen Sie optionale Filterung hinzu, definieren Sie die optionale Anreicherung und wählen Sie das Ziel für die Ereignisdaten.

Eine Quelle sendet Ereignisse an eine EventBridge Pipe, die passende Ereignisse filtert und an das Ziel weiterleitet.

Diese Integration ermöglicht es Ihnen, die Leistungsfähigkeit der Funktionen zur Analyse Timestream von Zeitreihendaten zu nutzen und gleichzeitig Ihre Datenerfassungspipeline zu vereinfachen.

Die Verwendung von EventBridge Pipes with Timestream bietet die folgenden Vorteile:

  • Datenaufnahme in Echtzeit: Streamen Sie Daten von Kinesis direkt zu Timestream für und ermöglichen so Analysen LiveAnalytics und Überwachung in Echtzeit.

  • Nahtlose Integration: Verwenden Sie EventBridge Pipes, um den Datenfluss zu verwalten, ohne dass komplexe benutzerdefinierte Integrationen erforderlich sind.

  • Verbesserte Filterung und Transformation: Filtern oder transformieren Sie Kinesis-Datensätze, bevor sie gespeichert werden Timestream , um Ihre spezifischen Datenverarbeitungsanforderungen zu erfüllen.

  • Skalierbarkeit: Verarbeiten Sie Datenströme mit hohem Durchsatz und sorgen Sie für eine effiziente Datenverarbeitung mit integrierten Parallelismus- und Batching-Funktionen.

Konfiguration

Gehen Sie wie folgt vor, um eine EventBridge Pipe einzurichten Timestream, an die Daten von Kinesis gestreamt werden sollen:

  1. Kinesis-Stream erstellen

    Stellen Sie sicher, dass Sie über einen aktiven Kinesis-Datenstream verfügen, aus dem Sie Daten aufnehmen möchten.

  2. Erstellen Sie eine Timestream Datenbank und eine Tabelle

    Richten Sie Ihre Timestream Datenbank und Tabelle ein, in der die Daten gespeichert werden.

  3. Konfigurieren Sie die EventBridge Pipe:

    • Quelle: Wählen Sie Ihren Kinesis-Stream als Quelle aus.

    • Ziel: Wählen Sie Timestream es als Ziel aus.

    • Batching-Einstellungen: Definieren Sie das Batch-Fenster und die Batchgröße, um die Datenverarbeitung zu optimieren und die Latenz zu reduzieren.

Wichtig

Beim Einrichten einer Pipe empfehlen wir, die Richtigkeit aller Konfigurationen zu testen, indem Sie einige Datensätze aufnehmen. Bitte beachten Sie, dass die erfolgreiche Erstellung einer Pipe nicht garantiert, dass die Pipeline korrekt ist und die Daten fehlerfrei fließen. Nach dem Anwenden der Zuordnung können Laufzeitfehler auftreten, wie z. B. eine falsche Tabelle, ein falscher dynamischer Pfadparameter oder ein ungültiger Timestream Datensatz, die entdeckt werden, wenn die tatsächlichen Daten durch die Pipe fließen.

Die folgenden Konfigurationen bestimmen die Geschwindigkeit, mit der Daten aufgenommen werden:

  • BatchSize: Die maximale Größe des Batches, für den Timestream gesendet wird. LiveAnalytics Bereich: 0 — 100. Es wird empfohlen, diesen Wert bei 100 zu belassen, um einen maximalen Durchsatz zu erzielen.

  • MaximumBatchingWindowInSeconds: Die maximale Wartezeit, bis der Stapel gefüllt ist, batchSize bevor der Stapel als LiveAnalytics Ziel an Timestream gesendet wird. Abhängig von der Rate eingehender Ereignisse entscheidet diese Konfiguration über die Verzögerung der Aufnahme. Es wird empfohlen, diesen Wert auf < 10 Sekunden zu belassen, um die Daten nahezu in Echtzeit zu Timestream senden.

  • ParallelizationFactor: Die Anzahl der Batches, die von jedem Shard gleichzeitig verarbeitet werden sollen. Es wird empfohlen, den Maximalwert 10 zu verwenden, um einen maximalen Durchsatz und eine Aufnahme nahezu in Echtzeit zu erzielen.

    Wenn Ihr Stream von mehreren Zielen gelesen wird, verwenden Sie den erweiterten Fan-Out, um einen speziellen Nutzer für Ihre Pipe bereitzustellen und so einen hohen Durchsatz zu erzielen. Weitere Informationen finden Sie im Kinesis Data Streams Benutzerhandbuch unter Entwickeln erweiterter Kinesis Data Streams API Fan-Out-Consumer mit dem.

Anmerkung

Der maximale Durchsatz, der erreicht werden kann, ist durch gleichzeitige Pipe-Ausführungen pro Konto begrenzt.

Die folgende Konfiguration stellt sicher, dass Datenverlust verhindert wird:

  • DeadLetterConfig: Es wird empfohlen, die Konfiguration immer so DeadLetterConfig zu konfigurieren, dass Datenverlust vermieden wird, wenn Ereignisse LiveAnalytics aufgrund von Benutzerfehlern nicht in Timestream aufgenommen werden konnten.

Optimieren Sie die Leistung Ihrer Pipe mit den folgenden Konfigurationseinstellungen, um zu verhindern, dass Aufzeichnungen zu Verlangsamungen oder Blockaden führen.

  • MaximumRecordAgeInSeconds: Ältere Datensätze werden nicht verarbeitet und direkt dorthin verschoben. DLQ Wir empfehlen, diesen Wert so einzustellen, dass er nicht höher ist als die konfigurierte Aufbewahrungsdauer für den Speicherspeicher der Timestream Zieltabelle.

  • MaximumRetryAttempts: Die Anzahl der Wiederholungsversuche für einen Datensatz, bevor der Datensatz an DeadLetterQueue gesendet wird. Es wird empfohlen, diesen Wert auf 10 zu konfigurieren. Dies sollte helfen, vorübergehende Probleme zu beheben. Bei anhaltenden Problemen wird der Datensatz in den Rest des Streams verschoben DeadLetterQueue und dort entsperrt.

  • OnPartialBatchItemFailure: Für Quellen, die eine teilweise Stapelverarbeitung unterstützen, empfehlen wir Ihnen, diese Option zu aktivieren und sie als AUTOMATIC _ BISECT zu konfigurieren, damit fehlgeschlagene Datensätze vor dem Ablegen/Senden an erneut versucht werden können. DLQ

Konfigurationsbeispiel

Hier ist ein Beispiel für die Konfiguration einer EventBridge Pipe zum Streamen von Daten aus einem Kinesis-Stream in eine Timestream Tabelle:

Beispiel IAM Richtlinien-Updates für Timestream
{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "timestream:WriteRecords"
            ],
            "Resource": [
                "arn:aws:timestream:us-east-1:123456789012:database/my-database/table/my-table"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "timestream:DescribeEndpoints"
            ],
            "Resource": "*"
        }
    ]
}
Beispiel Kinesis-Stream-Konfiguration
{
  "Source": "arn:aws:kinesis:us-east-1:123456789012:stream/my-kinesis-stream",
  "SourceParameters": {
    "KinesisStreamParameters": {
        "BatchSize": 100,
        "DeadLetterConfig": {
            "Arn": "arn:aws:sqs:us-east-1:123456789012:my-sqs-queue"
        },
       "MaximumBatchingWindowInSeconds": 5,
        "MaximumRecordAgeInSeconds": 1800,
        "MaximumRetryAttempts": 10,
        "StartingPosition": "LATEST",
       "OnPartialBatchItemFailure": "AUTOMATIC_BISECT"
    }
  }
}
Beispiel Timestream Zielkonfiguration
{
    "Target": "arn:aws:timestream:us-east-1:123456789012:database/my-database/table/my-table",
    "TargetParameters": {
        "TimestreamParameters": {
            "DimensionMappings": [
                {
                    "DimensionName": "sensor_id",
                    "DimensionValue": "$.data.device_id",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "DimensionName": "sensor_type",
                    "DimensionValue": "$.data.sensor_type",
                    "DimensionValueType": "VARCHAR"
                },
                {
                    "DimensionName": "sensor_location",
                    "DimensionValue": "$.data.sensor_loc",
                    "DimensionValueType": "VARCHAR"
                }
            ],
            "MultiMeasureMappings": [
                {
                    "MultiMeasureName": "readings",
                    "MultiMeasureAttributeMappings": [
                        {
                            "MultiMeasureAttributeName": "temperature",
                            "MeasureValue": "$.data.temperature",
                            "MeasureValueType": "DOUBLE"
                        },
                        {
                            "MultiMeasureAttributeName": "humidity",
                            "MeasureValue": "$.data.humidity",
                            "MeasureValueType": "DOUBLE"
                        },
                        {
                            "MultiMeasureAttributeName": "pressure",
                            "MeasureValue": "$.data.pressure",
                            "MeasureValueType": "DOUBLE"
                        }
                    ]
                }
            ],
            "SingleMeasureMappings": [],
            "TimeFieldType": "TIMESTAMP_FORMAT",
            "TimestampFormat": "yyyy-MM-dd HH:mm:ss.SSS",
            "TimeValue": "$.data.time",
            "VersionValue": "$.approximateArrivalTimestamp"
        }
    }
}

Transformation von Ereignissen

EventBridge Pipes ermöglichen es Ihnen, Daten zu transformieren, bevor sie ankommen. Timestream Sie können Transformationsregeln definieren, um die eingehenden Kinesis Datensätze zu ändern, z. B. um Feldnamen zu ändern.

Angenommen, Ihr Kinesis Stream enthält Temperatur- und Feuchtigkeitsdaten. Sie können diese Felder mithilfe einer EventBridge Transformation umbenennen, bevor Sie sie in Timestream einfügen.

Bewährte Methoden

Batching und Pufferung

  • Konfigurieren Sie das Batchfenster und die Größe so, dass ein ausgewogenes Verhältnis zwischen Schreiblatenz und Verarbeitungseffizienz besteht.

  • Verwenden Sie ein Batch-Fenster, um vor der Verarbeitung genügend Daten zu sammeln und so den Aufwand für häufige kleine Batches zu reduzieren.

Parallele Verarbeitung

Verwenden Sie diese ParallelizationFactorEinstellung, um die Parallelität zu erhöhen, insbesondere bei Streams mit hohem Durchsatz. Dadurch wird sichergestellt, dass mehrere Batches von jedem Shard gleichzeitig verarbeitet werden können.

Datentransformation

Nutzen Sie die Transformationsfunktionen von EventBridge Pipes, um Datensätze zu filtern und zu verbessern, bevor Sie sie speichern Timestream. Dies kann dazu beitragen, die Daten an Ihre analytischen Anforderungen anzupassen.

Sicherheit

  • Stellen Sie sicher, dass die für EventBridge Pipes verwendeten IAM Rollen über die erforderlichen Lese Kinesis - und Schreibberechtigungen verfügen. Timestream

  • Verwenden Sie Verschlüsselungs- und Zugriffskontrollmaßnahmen, um Daten während der Übertragung und Speicherung zu schützen.

Fehler beim Debuggen

  • Automatische Deaktivierung von Pipes

    Pipes werden in etwa 2 Stunden automatisch deaktiviert, wenn das Ziel nicht existiert oder es Probleme mit den Berechtigungen gibt

  • Drosselungen

    Pipes sind in der Lage, sich automatisch zurückzuziehen und es erneut zu versuchen, bis die Drosselung reduziert ist.

  • Protokolle aktivieren

    Wir empfehlen Ihnen, Logs auf ERROR Ebene zu aktivieren und Ausführungsdaten einzubeziehen, um mehr Einblicke in Fehlschläge zu erhalten. Bei einem Fehler enthalten diese Protokolle request/response sent/received von Timestream. Dies hilft Ihnen, den damit verbundenen Fehler zu verstehen und die Datensätze nach der Behebung gegebenenfalls erneut zu verarbeiten.

Überwachen

Wir empfehlen Ihnen, Alarme für Folgendes einzurichten, um Probleme mit dem Datenfluss zu erkennen:

  • Maximales Alter des Datensatzes in der Quelle

    • GetRecords.IteratorAgeMilliseconds

  • Messwerte für Fehler in Pipes

    • ExecutionFailed

    • TargetStageFailed

  • Timestream APIFehler schreiben

    • UserErrors

Weitere Monitoring-Metriken finden Sie unter Überwachung EventBridge im EventBridge Benutzerhandbuch.