新規でWEBサービスを作るなら採用するデータ連携を考慮したアーキテクチャ

dataContractSpecification: 0.9.3
id: datastream_sample

info:
  ...

servers:
  production:
    type: s3
    location: s3://...
    format: csv

models:
  todos:
    description: All of Todos"
    type: table
    fields:
      id:
        type: long
        description: "ID"
        nullable: false
        primaryKey: true
      title:
        type: string
        description: "Title"
        minLength: 1
        maxLength: 100
      completed:
        type: boolean
        description: "Completed"
        nullable: true
      description:
        type: string
        description: "Description"
        nullable: true
quality:
  type: SodaCL
  specification:
    checks for todos:
      - row_count > 1
      - duplicate_count(id) = 0

datacontract export --format html datacontract-s3.yaml > datacontract-s3.html

$ datacontract test datacontract-s3.yaml 
Testing datacontract-s3.yaml
╭────────┬────────────────────────────────────────────────┬───────┬─────────╮
│ Result │ Check                                          │ Field │ Details │
├────────┼────────────────────────────────────────────────┼───────┼─────────┤
│ passed │ Check that field id is present                 │       │         │
│ passed │ Check that field title is present              │       │         │
│ passed │ Check that field completed is present          │       │         │
│ passed │ Check that field description is present        │       │         │
│ passed │ row_count > 1                                  │       │         │
│ passed │ Check that field title has a min length of 1   │ title │         │
│ passed │ Check that field title has a max length of 100 │ title │         │
│ passed │ duplicate_count(id) = 0                        │ id    │         │
╰────────┴────────────────────────────────────────────────┴───────┴─────────╯
🟢 data contract is valid. Run 8 checks. Took 1.120317 seconds.

{
  "schema": // ...,
  "payload": {
    "before": null,
    "after": // モデルの中身,
  }
}

{
  "name": "postgres-connector",
  "config": {
    ...
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": false,
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": false
  }
}

openapi: 3.0.0
info:
  title: Todo Service
  version: 1.0.0
  description: A service for managing todo items

paths: {}

components:
  schemas:
    CreateTodoEvent:
      description: |
        Topic: postgres.public.CreateTodoEvent
        Todoが作成されたら作成されるイベント
      type: object
      required:
        - hash
        - todo_id
        - title
        - description
      properties:
        hash:
          type: string
          format: uuid
          description: Unique identifier for the todo item
        todo_id:
          type: integer
          format: int64
          description: Unique identifier for the todo item
        title:
          type: string
          description: Title of the todo item
        description:
          type: string
          description: Detailed description of the todo item

npx openapi-zod-client openapi-event.yaml -o src/event/event.ts --export-schemas

import { Kafka } from "kafkajs";
import { z } from "zod";
import { schemas } from "./event/event";

const kafka = new Kafka({
  clientId: "todo-event-consumer",
  brokers: ["localhost:9092"],
});

const consumer = kafka.consumer({ groupId: "todo-event-group" });

interface EventEnvelope<T> {
  before?: T;
  after?: T;
}

const EventEnvelopeSchema = <T extends z.ZodTypeAny>(schema: T) =>
  z.object({
    before: schema.nullable().optional(),
    after: schema.nullable().optional(),
  });

const parseEventEnvelope = <T extends z.ZodTypeAny>(
  value: string,
  schema: T
): EventEnvelope<z.infer<T>> => {
  const parsed = JSON.parse(value);
  return EventEnvelopeSchema(schema).parse(parsed);
};

// 以下のコードを追加
const run = async () => {
  await consumer.connect();
  await consumer.subscribe({
    topic: "postgres.public.CreateTodoEvent",
    fromBeginning: true,
  });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      try {
        const createTodoEvent = parseEventEnvelope(
          message.value?.toString() || "{}",
          schemas.CreateTodoEvent
        );
        console.log({
          hash: createTodoEvent.after?.hash,
          todo_id: createTodoEvent.after?.todo_id,
          title: createTodoEvent.after?.title,
          description: createTodoEvent.after?.description,
        });
      } catch (error) {
        console.error("メッセージのパースに失敗しました:", error);
      }
    },
  });
};

run().catch(console.error);

dataContractSpecification: 0.9.3
...

servers:
  production:
    type: bigquery
...

models:
  public_Todo:
    description: All of Todos"
    type: table
    servicelevels:
      freshness:
        interval: 1d
        freshness: 1d
    fields:
      id:
        type: long
        description: "ID"
        nullable: false
        primaryKey: true
      title:
        type: string
        description: "Title"
        minLength: 1
        maxLength: 100
      completed:
        type: boolean
        description: "Completed"
        nullable: true
      description:
        type: string
        description: "Description"
        nullable: true

datacontract export --format dbt datacontract.yaml > dbt/models/schema.yaml

datacontract export --format dbt-sources datacontract.yaml > dbt/models/sources.yaml

npx swagger-typescript-api -p openapi.yaml -o src/server --no-client --route-types --extract-response-error --type-prefix Api -n openapi.ts --responses

npx openapi-zod-client ./openapi.yaml -o ./src/client/client.ts"