À utiliser PutRecords avec un AWS SDK ou CLI

Pour écrire plusieurs enregistrements dans un flux de données

L'put-recordsexemple suivant écrit un enregistrement de données à l'aide de la clé de partition spécifiée et un autre enregistrement de données à l'aide d'une clé de partition différente en un seul appel.

aws kinesis put-records \
    --stream-name samplestream \
    --records Data=blob1,PartitionKey=partitionkey1 Data=blob2,PartitionKey=partitionkey2

Sortie :

{
    "FailedRecordCount": 0,
    "Records": [
        {
            "SequenceNumber": "49600883331171471519674795588238531498465399900093808706",
            "ShardId": "shardId-000000000004"
        },
        {
            "SequenceNumber": "49600902273357540915989931256902715169698037101720764562",
            "ShardId": "shardId-000000000009"
        }
    ],
    "EncryptionType": "KMS"
}

Pour plus d'informations, consultez la section Developing Producers Using the Amazon Kinesis Data API Streams with AWS SDK the for Java dans le manuel du développeur Amazon Kinesis Data Streams.

import { PutRecordsCommand, KinesisClient } from "@aws-sdk/client-kinesis";

/**
 * Put multiple records into a Kinesis stream.
 * @param {{ streamArn: string }} config
 */
export const main = async ({ streamArn }) => {
  const client = new KinesisClient({});
  try {
    await client.send(
      new PutRecordsCommand({
        StreamARN: streamArn,
        Records: [
          {
            Data: new Uint8Array(),
            /**
             * Determines which shard in the stream the data record is assigned to.
             * Partition keys are Unicode strings with a maximum length limit of 256
             * characters for each key. Amazon Kinesis Data Streams uses the partition
             * key as input to a hash function that maps the partition key and
             * associated data to a specific shard.
             */
            PartitionKey: "TEST_KEY",
          },
          {
            Data: new Uint8Array(),
            PartitionKey: "TEST_KEY",
          },
        ],
      }),
    );
  } catch (caught) {
    if (caught instanceof Error) {
      //
    } else {
      throw caught;
    }
  }
};

// Call function if run directly.
import { fileURLToPath } from "node:url";
import { parseArgs } from "node:util";

if (process.argv[1] === fileURLToPath(import.meta.url)) {
  const options = {
    streamArn: {
      type: "string",
      description: "The ARN of the stream.",
    },
  };

  const { values } = parseArgs({ options });
  main(values);
}