RecordsSink

Source: src/AWS/Timestream/RecordsSink.ts

A batching sink over Timestream WriteRecords (100 records per call).

Timestream ingests the valid subset of each batch and reports invalid records positionally via RejectedRecordsException (schema conflicts, timestamps outside the retention window, version conflicts). Rejections are permanent — the sink drops them (surfacing a warning with the rejected count) and keeps draining; there is no transient per-record failure mode to retry.

Provide Timestream.RecordsSinkHttp on the Function to implement the binding.

Streaming Records

// init — bind the sink to the table; shared attributes are sent once per
// batch as CommonAttributes and merged into every record server-side
const sink = yield* Timestream.RecordsSink(table, {
  commonAttributes: {
    MeasureName: "cpu",
    MeasureValueType: "DOUBLE",
    TimeUnit: "MILLISECONDS",
  },
});

// runtime — drain a stream through the sink (batched 100 records/call)
yield* Stream.fromIterable(
  samples.map((s) => ({
    Dimensions: [{ Name: "host", Value: s.host }],
    MeasureValue: `${s.value}`,
    Time: `${s.time}`,
  })),
).pipe(Stream.run(sink));