Data Ingestion and Processing Pipeline

**Referenced Files in This Document** - [[main.go]](file/bi-api-jushuitan/cmd/bi-api-jushuitan/main.go) - [[main.go]](file/bi-api-leke/cmd/bi-api-leke/main.go) - [[config.go]](file/bi-common/mq/kafkax/config.go) - [[producer.go]](file/bi-common/mq/kafkax/producer.go) - [[consumer.go]](file/bi-common/mq/kafkax/consumer.go) - [[consumer_manager.go]](file/bi-common/mq/kafkax/consumer-manager.go) - [[multi_consumer.go]](file/bi-common/mq/kafkax/multi-consumer.go) - [[retry.go]](file/bi-common/mq/kafkax/retry.go) - [[auth.go]](file/bi-api-jushuitan/common/sdkhelper/auth.go) - [[streamload.go]](file/bi-api-jushuitan/internal/data/client/streamload.go) - [[client.go]](file/bi-common/database/starrocks/streamload/client.go) - [[response.go]](file/bi-common/database/starrocks/streamload/response.go) - [[simulation_test.go]](file/bi-common/database/starrocks/streamload/simulation-test.go) - [[cost_erp.go]](file/bi-basic/internal/biz/logic/cost-erp.go) - [[PROJECT_REPORT.md]](file/bi-api-jushuitan/project-report.md) - [[root.go]](file/bi-cron/cmd/root.go)

Table of Contents

1. Introduction
2. Project Structure
3. Core Components
4. Architecture Overview
5. Detailed Component Analysis
6. Dependency Analysis
7. Performance Considerations
8. Troubleshooting Guide
9. Conclusion

Introduction

This document describes the data ingestion and processing pipeline that integrates external ERP systems (Leke ERP and Jushuitan ERP), streams real-time events via Apache Kafka, transforms and normalizes data, and loads it into StarRocks for analytics. It covers authentication with external APIs, Kafka producer and consumer configurations, message serialization, transformation to normalized tables, batch historical loads, streaming real-time updates, error handling, retries, validation, consistency guarantees, idempotency, and failure recovery.

Project Structure

The pipeline spans several modules:

• bi-api-jushuitan: HTTP/gRPC server, Kafka consumer manager, SDK helper for Jushuitan ERP, and StarRocks StreamLoad integration.
• bi-api-leke: Kratos application entrypoint with Kafka consumer manager.
• bi-common: Shared Kafka client library, StarRocks StreamLoad client, and common configuration structures.
• bi-basic: Business logic for ERP account management and related entities.
• bi-cron: Scheduled job runner for historical data loads.

Diagram sources

• [PROJECT_REPORT.md]
• [main.go]
• [main.go]

Section sources

• [PROJECT_REPORT.md]

Core Components

• Kafka Producer/Consumer Library: Provides typed configuration, batching, compression, retries, and worker pools for consumption.
• Jushuitan SDK Helper: Implements OAuth-like token exchange and authorization URL building for Jushuitan ERP.
• StarRocks StreamLoad Client: Handles HTTP-based StreamLoad with authentication, retry, and response parsing.
• Bi-API-Jushuitan: Orchestrates Kafka consumers and writes normalized records to StarRocks via StreamLoad.
• Bi-API-Leke: Kratos-based entrypoint wiring Kafka consumers.
• Bi-Basic: Manages ERP accounts and related metadata used by ingestion services.
• Bi-Cron: Scheduled runner for historical data loads.

Section sources

• [config.go]
• [producer.go]
• [consumer.go]
• [auth.go]
• [client.go]
• [streamload.go]

Architecture Overview

The ingestion pipeline follows a publish-subscribe pattern:

• External ERP systems push change events to Kafka topics.
• Kafka consumers (per service) subscribe to topics and process messages.
• Messages are transformed into normalized StarRocks table rows.
• StarRocks StreamLoad ingests JSON payloads into target tables.

Diagram sources

• [multi_consumer.go]
• [consumer.go]
• [client.go]

Detailed Component Analysis

Kafka-Based Ingestion Architecture

• Producer configuration supports batching, compression, acknowledgments, retries, timeouts, and SASL/TLS.
• Consumer configuration supports group-based partition assignment, isolation level, heartbeat/session timeouts, and partition change monitoring.
• Consumer Manager provisions multiple consumers with worker pools and optional global handler registration.
• Multi-Topic Consumer enables consuming from multiple topics concurrently with shared settings.

Diagram sources

• [config.go]
• [consumer_manager.go]

Section sources

• [config.go]
• [consumer_manager.go]
• [multi_consumer.go]

API Authentication with Jushuitan ERP

• Builds authorization URLs and exchanges authorization codes for tokens.
• Supports refresh tokens and signing requests.
• Used by bi-api-jushuitan to obtain credentials for ERP API calls.

Diagram sources

• [auth.go]

Section sources

• [auth.go]

Data Transformation and StarRocks StreamLoad

• Records are normalized and serialized to JSON.
• StreamLoad client posts JSON to StarRocks FE endpoint with Basic auth and label.
• Response includes status, row counts, and timing metrics; success determined by status field.

Diagram sources

• [streamload.go]
• [client.go]
• [response.go]

Section sources

• [streamload.go]
• [client.go]
• [response.go]

Batch Historical Loads and Streaming Updates

• Historical loads are orchestrated by bi-cron, scheduled via Kubernetes CronJob.
• Real-time updates are handled by Kafka consumers started in bi-api-jushuitan and bi-api-leke entrypoints.

Diagram sources

• [root.go]
• [main.go]

Section sources

• [root.go]
• [main.go]

Error Handling, Retries, and Validation

• Kafka producers support retries with exponential backoff and jitter.
• Consumers expose commit/ack controls and lag metrics; graceful shutdown supported.
• StreamLoad responses define success/failure semantics; tests simulate Nacos-backed configuration loading.

Diagram sources

• [retry.go]
• [producer.go]

Section sources

• [retry.go]
• [consumer.go]
• [simulation_test.go]

Data Consistency Guarantees and Idempotency

• Kafka consumer groups with committed offsets enable at-least-once delivery; idempotency depends on downstream deduplication.
• StreamLoad labels prevent duplicate ingestion of the same payload; success/failure semantics guide retry decisions.
• Snowflake IDs ensure uniqueness during normalization.

Section sources

• [consumer.go]
• [client.go]
• [streamload.go]

Dependency Analysis

• bi-api-jushuitan depends on bi-common for Kafka and StarRocks clients.
• Both bi-api services depend on bi-common for shared configuration and messaging primitives.
• bi-basic provides entity definitions and business logic for ERP account management.

Diagram sources

• [main.go]
• [main.go]

Section sources

• [main.go]
• [main.go]

Performance Considerations

• Producer batching and compression reduce network overhead; tune BatchSize/BatchBytes/BatchTimeout per throughput targets.
• Consumer worker pools parallelize message processing; adjust worker count based on CPU and IO capacity.
• StreamLoad timeouts and retry limits should align with SLAs; monitor response timings for bottlenecks.
• Partitioning strategy should match topic volume and consumer scaling needs.

Troubleshooting Guide

• Kafka connectivity: Use Ping helpers on producer/consumer to validate broker reachability.
• Consumer lag: Monitor Lag() and ReadLagInterval settings to detect backlog.
• StreamLoad failures: Inspect Response fields (Status, Message, ErrorURL) and re-ingest with unique labels.
• Configuration loading: Verify Nacos-backed configs for StarRocks and Kafka; fallback defaults are embedded.

Section sources

• [consumer.go]
• [producer.go]
• [response.go]
• [simulation_test.go]

Conclusion

The pipeline integrates external ERP systems with a robust Kafka-based ingestion architecture, enabling scalable real-time processing and reliable historical loads into StarRocks. With structured authentication, typed Kafka configurations, explicit transformation and validation, and comprehensive error handling/recovery, the system supports high-throughput, observable, and maintainable data operations.