131 lines
5.1 KiB
YAML
131 lines
5.1 KiB
YAML
# streams/transaction.yaml
|
|
input:
|
|
nats_jetstream:
|
|
#urls: [ "nats-cluster:4222" ]
|
|
urls: [ "nats://nats-cluster.infra.svc.cluster.local:4222" ]
|
|
stream: "request_stream"
|
|
subject: "input_transaction_logs"
|
|
manage_stream: true # <--- Add this
|
|
|
|
# NEW: Global Pipeline for the input
|
|
pipeline:
|
|
processors:
|
|
- mapping: |
|
|
# Set metadata globally so all outputs and templates can see it
|
|
root = this
|
|
meta conflict_col = this.data.order.id.or("unknown")
|
|
|
|
# with thhe following assignment only streams can see the value but template not
|
|
meta transaction_id = this.transaction_id.or("unknown")
|
|
meta transaction_status = this.transaction_status
|
|
root.timestamp = this.timestamp
|
|
meta order_id = this.data.order.id.or("unknown")
|
|
# root.flat_status = this.data.order.status.lowercase()
|
|
# root.order_customer_email = this.data.order.customer.email
|
|
# root.order_customer_loyalty_tier = this.data.order.customer.loyalty_tier
|
|
# root.metadata_ip_address = this.data.metadata.ip_address
|
|
# root.metadata_region = this.data.metadata.region
|
|
# root.metadata_user_agent = this.data.metadata.user_agent
|
|
root.data.order.items = this.data.order.items
|
|
# meta metadata_items = this.data.items.map_each(item -> item.sku)
|
|
|
|
- log:
|
|
level: INFO
|
|
message: "New transaction: status ${! meta(\"transaction_status\") } - order ID: ${! meta(\"order_id\") } timestamp: ${! json(\"timestamp\")} timestamp2a ${! this.timestamp }"
|
|
|
|
# output:
|
|
# broker:
|
|
# pattern: fan_out
|
|
# outputs:
|
|
# # --- Branch 1: Archive the raw JSON ---
|
|
# - yugabyte_insert:
|
|
# dsn: "postgres://yugabyte@yugabytedb-3:5433/ecommerce?sslmode=disable"
|
|
# table_name: "public.transaction"
|
|
# #conflict_columns: "transaction_id"
|
|
# #columns: ["transaction_id", "orders", "order_status"]
|
|
# # if use this deactivate conflict_columns
|
|
# columns: ["content_hash", "orders", "status","created_at"]
|
|
# value_mapping: |
|
|
# root = [
|
|
# # if use meta define it first above
|
|
# # meta("transaction_id"),
|
|
# # meta("order_id"),
|
|
# # meta("order_status")
|
|
# this.transaction_id,
|
|
# this.data.order.id,
|
|
# this.transaction_status,
|
|
# this.timestamp
|
|
# ]
|
|
|
|
# # --- Branch 2: Extract nested GraphQL data into Facts ---
|
|
|
|
# - yugabyte_insert:
|
|
# dsn: "postgres://yugabyte@yugabytedb-3:5433/ecommerce?sslmode=disable"
|
|
# table_name: "public.order"
|
|
# # conflict_columns: "content_hash"
|
|
# columns: ["content_hash", "customer_email", "customer_loyalty_tier","status", "ip_address","region"]
|
|
# # if use this deactivate conflict_columns
|
|
# #columns: ["content_hash", "customer_email", "status"]
|
|
# value_mapping: |
|
|
# root = [
|
|
# # if use meta define it first above
|
|
# # meta("order_id"),
|
|
# # meta("order_customer_email"),
|
|
# # meta("order_status")
|
|
# this.data.order.id,
|
|
# this.data.order.customer.email,
|
|
# this.data.order.customer.loyalty_tier,
|
|
# this.data.order.status,
|
|
# this.data.order.metadata.ip_address,
|
|
# this.data.order.metadata.region
|
|
# ]
|
|
|
|
# - yugabyte_insert:
|
|
# dsn: "postgres://yugabyte@yugabytedb-3:5433/ecommerce?sslmode=disable"
|
|
# table_name: "public.items"
|
|
# columns: ["content_hash", "items"]
|
|
# # No unarchive needed!
|
|
# processors:
|
|
# - mapping: |
|
|
# root = this
|
|
# # 1. Update dedupe key for this specific table
|
|
# meta conflict_col = meta("order_id") + "-item"
|
|
|
|
# # 2. Stringify the items array for the JSONB column
|
|
# # root.items_blob = this.data.order.items.format_json()
|
|
|
|
# value_mapping: |
|
|
# root = [
|
|
# meta("order_id"),
|
|
# this.data.order.items.string()
|
|
# ]
|
|
|
|
# - kafka:
|
|
# addresses: [ "redpanda:9092" ]
|
|
# topic: "transactions_redpanda"
|
|
# # By not adding a 'processors' block here,
|
|
# # it sends the current state of 'root' from the global pipeline.
|
|
# # To ensure it is the ABSOLUTE original, we can force a mapping:
|
|
# key : ${! meta("transaction_id")}
|
|
# processors:
|
|
# - mapping: |
|
|
# root = content()
|
|
|
|
# 1. example below for using content_hash
|
|
# 2. change it accordingly to your defined conflict_columns
|
|
# 3. IMPORTANT: conflict_columns is to be defined as UNIQUE in sql
|
|
# CREATE TABLE public.transaction (
|
|
# id BIGSERIAL,
|
|
# content_hash TEXT UNIQUE,
|
|
# orders TEXT,
|
|
# order_status TEXT,
|
|
# timestamp TIMESTAMPTZ,
|
|
# PRIMARY KEY (id)
|
|
# );
|
|
|
|
# CREATE TABLE public.order (
|
|
# id TEXT PRIMARY KEY, -- Must be TEXT for "ORD-5521"
|
|
# content_hash TEXT UNIQUE,
|
|
# customer_email TEXT,
|
|
# amount DECIMAL
|
|
# ); |