Clojure 数据管道与流处理：Kafka 消费、异步管道与 ETL

;; 管道骨架：source → 变换函数 → sink
(defn pipeline [source-fn transform-fn sink-fn]
  (let [in  (async/chan 100)
        out (async/chan 100)]
    (async/go-loop []
      (when-let [x (async/<! in)]
        (async/>! out (transform-fn x))
        (recur)))
    (async/go-loop []
      (when-let [x (async/<! out)]
        (sink-fn x)
        (recur)))
    {:in in :out out}))

(require '[clj-kafka.producer :as producer])

(def producer-config
  {"bootstrap.servers" "localhost:9092"
   "key.serializer"    "org.apache.kafka.common.serialization.StringSerializer"
   "value.serializer"  "org.apache.kafka.common.serialization.StringSerializer"
   "acks"              "all"})          ; all = 强一致

(def producer (producer/creator producer-config))

;; 发送业务事件
(producer/send producer
               {:topic "order.events"
                :key   (str order-id)
                :value (json/write-str {:order_id order-id :amount 100})})

(require '[clj-kafka.consumer :as consumer])

(def consumer-config
  {"bootstrap.servers"  "localhost:9092"
   "group.id"           "order-analytics"
   "key.deserializer"   "org.apache.kafka.common.serialization.StringDeserializer"
   "value.deserializer" "org.apache.kafka.common.serialization.StringDeserializer"
   "enable.auto.commit" "false"})   ; 手动提交，配合幂等处理

(defn consume! []
  (consumer/with-consumer [consumer consumer-config]
    (consumer/subscribe consumer #{:order.events})
    (doseq [record (consumer/poll consumer 1000)]
      (process-event! (json/parse-string (:value record) true))
      ;; 处理成功后提交位移，失败则不提交（会重投递）
      (consumer/commit! consumer))))

;; 通道作为背压缓冲：限制在途消息数
(def queue (async/chan 200))   ; 最多 200 条在途

(defn feeder [topic]
  (async/go-loop []
    (when-let [msg (async/<! (kafka-poll! topic))]
      (async/>! queue msg)      ; 队列满则阻塞（背压传导到 poll）
      (recur))))

(defn worker [queue]
  (async/go-loop []
    (when-let [msg (async/<! queue)]
      (process-msg! msg)
      (recur))))

;; 启动 N 个 worker 并行消费，总数由通道缓冲限制
(dotimes [i 8] (worker queue))

topic: order.events（3 分区）
消费组 order-analytics（3 个消费者实例）
分区 0 → 消费者 A    分区 1 → 消费者 B    分区 2 → 消费者 C
key = order_id → 同一订单的事件总是进同一分区 → 顺序有保证

;; 标准 ETL 变换函数：净化 → 规范化 → 富化
(defn clean-record [raw]
  (-> raw
      (select-keys [:order_id :amount :paid_at :user_id])
      (update :amount (fn [v] (if (string? v) (parse-double v) v)))
      (update :paid_at #(when % (str->instant %)))
      (assoc :received_at (System/currentTimeMillis))))

(defn enrich-record [rec]
  (assoc rec :amount_tier (tier-of (:amount rec))
             :region (lookup-region (:user_id rec))))

;; 组装 ETL pipeline
(defn etl-pipeline [source-chan sink-fn]
  (->> source-chan
       (map clean-record)
       (filter valid?)
       (map enrich-record)
       (run! sink-fn)))

;; 落库用「upsert」保证幂等：重放也不重复
(defn sink-to-db [db rec]
  (jdbc/execute! db
    ["insert into order_analytics (order_id, amount, amount_tier, region)
      values (?, ?, ?, ?)
      on conflict (order_id) do update
      set amount = excluded.amount,
          amount_tier = excluded.amount_tier"
     (:order_id rec) (:amount rec) (:amount_tier rec) (:region rec)]))

;; 按分钟窗口聚合指标（用 partition-by + reduce）
(defn windowed-aggregate [events window-ms]
  (->> events
       (map (juxt :ts identity))
       (partition-by (fn [[ts _]] (/ ts window-ms)))
       (map (fn [batch]
              {:window (first batch)
               :total  (reduce + (map (comp :amount second) batch))
               :count  (count batch)}))))

;; 用去重表保证幂等：记录已处理的 event_id
(defn process-idempotent [db event]
  (if (processed? db (:event_id event))
    :already-processed                     ; 幂等跳过
    (do (jdbc/transaction db
          (sink-to-db db event)
          (mark-processed! db (:event_id event)))
        :processed)))

;; 暴露管道指标（Prometheus）
(prom/gauge :pipeline_backlog "管道积压" (fn [] (async/count queue)))