OpenRec Real-Time Data Processor

Release v0.2.0 · Changelog

data-processor provides equivalent Flink and Spark Structured Streaming jobs. Both consume the user, item, and event Kafka topics, update Redis serving data, persist the original entities in HBase, and append immutable JSON records to Hive-backed HDFS locations for offline training.

Pipeline organization

Both engines use source/, process/, sink/, and job/ packages. Sources expose native DataStream<String> (Flink) or a Dataset<Row> with a json column (Spark); jobs accept these streams independently of their source. Add another source by producing the same input contract. Spark's HistoryEntitySource supplies bounded mutation history for recovery.

• EntityProjectionJob assembles entity serving projection and durable history writes.
• FeatureAggregationJob assembles event updates, engine-managed state, and snapshot sinks.
• Spark's FeatureRebuildJob coordinates the existing startup recovery from durable history, including refreshing inactive entities. It is part of startup, not a separate submitted job.

feature-core owns engine-neutral feature computation and process.ServingProjection owns shared user/item/event serving rules. Its Store port isolates these rules from Redis clients; engine adapters own connections and the existing Redis mutation fence. Engine-specific state stays in Flink's EventFeatureProcessFunction and Spark's FeatureAggregationProcess. Storage-specific sinks own Redis, HBase, and HDFS writes. Spark composite sinks retain one foreachBatch boundary and the existing history-before-serving write order; they do not provide a transaction across stores.

DpJob and SparkFeatureJob remain the deployment entry points. Configuration keys, topic/group selection, operator ordering, state descriptor, checkpoint paths, output schemas and destinations are unchanged. This classification does not split production deployment into independently consuming jobs. Use the existing parity/operator tests and cluster acceptance flows when changing processing rules or adding connectors; checkpoint restore from a real deployment still needs a savepoint/checkpoint rehearsal.

Feature Contract

The canonical feature registry is model/feature/catalog/feature.catalog.json; its packaged copy is feature-core/src/main/resources/openrec-feature-catalog.json. Synchronize it with the model repository's publish_catalog.py and use --check to detect drift. Model training selects subsets of these global features; selecting a subset does not change the streaming producer configuration. New features require matching implementations here and in rec-algorithm, plus any required historical backfill, before rec-console can offer them for training and deployment.

Feature formulas live in feature-core; Flink and Spark only supply engine-specific state and sinks. Raw user fields (id, device/profile/location/tags and register/login time) and item fields (id, title/category/subcategory/tags/scene, lifecycle/status and weight) are retained unchanged. Title, subcategory, tags and publication time are the online source fields for the content encoders shared by rec-algorithm and rank-engine; the stream layer preserves them without lossy preprocessing. Event streams generate the same behavioral columns as rec-algorithm/algorithm/feature/event_feature.py:

• totals: event_count, value sum/mean, active days, unique scenes and counterpart count;
• time: first/last event, recency, and 1/7/30-day counts;
• actions: click, expose, buy, collect and stay counts, plus the legacy labelled-event click rate;
• conversion: standard CTR (click / expose), collect per click, buy per click and buy per collect, each over lifetime and inclusive 1/7/30-day event-time windows.
• commerce: weighted category/subcategory preferences and per-action, 1/7/30-day price statistics from the immutable _openrecItemContext attached to each event.

Each event updates both its user and item snapshot. Redis keys are feature:user:{id} and feature:item:{id}. Raw serving keys remain compatible with rec-server (user:{id}, item:{id}, event:{userId}:scene:type, and new:{scene}). A structured dislike value is expanded into id:, category:, and one or more tag: members so BlackNode can apply it online; other event members remain item IDs. new:{scene} is a sorted set scored by pubTime; redis.new.max-items bounds every scene so the realtime projection cannot grow without limit.

Durable Training Data

Both jobs preserve the Kafka JSON byte-for-byte in HBase tables openrec_user, openrec_item, and openrec_event, under column entity:json. User and item ids are row keys; events use eventId, falling back to time#userId#itemId#scene#type#traceId. traceId is an attribution context and is not assumed to identify one action. Tables are created idempotently when a task starts.

The same payloads are appended under hdfs://namenode:8020/openrec/hive/{user,item,event}/dt=YYYY-MM-DD. The UTC partition date comes from event time, item modifyTime/pubTime, or user loginTime/registerTime; malformed or missing timestamps fall back to the processing date. Install the partitioned external Hive tables once after the cluster starts:

docker exec -i hiveserver2 /opt/hive/bin/beeline \
  -u jdbc:hive2://hiveserver2:10000 -n hive \
  -f /opt/workspace/data-processor/sql/openrec_entities.sql

If the source tree is not mounted at /opt/workspace, copy the SQL file into the container first. Offline embedding, i2i, and hot training jobs should read openrec.user_entity, openrec.item_entity, and openrec.event_entity, parse the JSON fields they require, and publish serving outputs to Redis/Elasticsearch. Feature snapshots remain under /openrec/features. Checkpoints are stored separately; never use checkpoint files as training input.

The DDL changed from an unpartitioned table to PARTITIONED BY (dt STRING). Drop and recreate an older development table before deploying this version (external data is not deleted). Scheduled algorithm jobs register only their requested day with ALTER TABLE ADD IF NOT EXISTS PARTITION.

Kafka messages use the version 1 mutation envelope published by rec-server: entity type, INSERT/UPDATE/DELETE operation, millisecond occurredAt, and the entity payload. Both processors also accept legacy bare-entity JSON as INSERT during the compatibility window. INSERT and UPDATE are upserts. User and item deletes remove Redis serving state and append tombstones to historical storage so cumulative offline readers do not resurrect deleted entities. Event mutations use eventId as their stable identity (with a deterministic field-based fallback), ignore older out-of-order mutations, and let DELETE win an equal-occurredAt tie. Event tombstones retract the event from Redis aggregates while remaining in append-only history for point-in-time offline resolution. See rec-proto for the shared mutation contract.

Build and Run

Java formatting

All Java production and test sources use the vendored Alibaba Eclipse profile through Spotless. With JDK 21 and Maven on PATH, run once per clone:

bash scripts/install-git-hooks.sh

Format and validate all three modules from the repository root:

bash scripts/java-format.sh       # applies formatting (mvn spotless:apply)
bash scripts/java-format.sh check # checks formatting (mvn spotless:check)

OPENREC_JAVA21_HOME can select JDK 21; MVN can select a Maven executable. Normal Maven builds also check formatting during validate and fail on differences. The pre-commit hook checks an isolated copy of the staged tree, and pre-push checks each outgoing tip. Hooks never stage files or overwrite partially staged changes. If a check fails, run the formatter, review and stage the result, then retry. New clones must install hooks; Git does not activate tracked hooks automatically.

CI runs Java format on pull requests and pushes to every branch, independently of dependency builds and unit tests. Configure GitHub branch protection/rulesets to require Java format and pull requests, with bypasses disabled, to prevent unformatted changes from merging. Local hooks can be bypassed; CI runs after a push and cannot by itself prevent objects from reaching the remote repository.

Use JDK 21:

mvn clean test
mvn -pl flink -am -DskipTests package
mvn -pl spark -am -DskipTests package

Submit one implementation for production, using the properties bundled in its jar:

docker cp flink/target/rec-flink-0.2.0.jar \
  flink-jobmanager:/opt/flink/jobs/openrec-features.jar
docker exec flink-jobmanager flink run -d -c com.openrec.dp.flink.DpJob \
  /opt/flink/jobs/openrec-features.jar
spark-submit --class com.openrec.dp.spark.SparkFeatureJob \
  --master spark://spark-master:7077 spark/target/rec-spark-0.2.0.jar

Configure Kafka, Redis, HBase, Hive/HDFS, checkpoint paths, parallelism, and event lateness in each module's src/main/resources/dp.properties, or pass an external properties file as the first application argument to override defaults. Set hbase.enabled=false or hive.enabled=false only when intentionally running without that cluster component. Use distinct Kafka consumer groups and checkpoint directories when comparing engines. Running both against the same topics duplicates persisted entities, although stable HBase row keys make user/item updates idempotent.

Delivery and compatibility

The two engines implement the same projection contract but are alternative deployments, not an active/active pair. Their default consumer groups are distinct, so starting both processes every message twice. Redis user/item writes and HBase user/item row keys are idempotent; event history and Hive/HDFS output are append-oriented and can contain duplicates after retries or dual-engine runs. Offline readers must collapse mutations by entity id and event time and apply the latest DELETE tombstone.

Flink checkpoints every 60 seconds and allows 30 seconds of event-time disorder by default. Spark uses the HDFS checkpoint path configured in dp.properties. Preserve or deliberately migrate these locations during an engine upgrade; checkpoint files are recovery state, not training input.

Spark feature state now uses features-json-v1-catalog-<version> beneath that root. The old features Kryo checkpoint is retained, since changing feature classes can make its binary state unreadable only when an old entity next receives an event. On first start without a committed checkpoint in the new namespace, the job reconstructs state from all durable hdfs.output/hive/event mutation history, refreshes inactive Redis feature snapshots, then replays retained Kafka events with the same identity/tombstone deduplication. Raw user/item/event checkpoints are unchanged. Recovery requires complete history; a legacy checkpoint with no history fails explicitly. Keep history and old checkpoints for rollback. Bump the JSON state namespace when making incompatible accumulator changes, even if the catalog version is unchanged.

This repository currently builds Java 21 bytecode against Flink 2.2.1, Spark 4.0.4/Scala 2.13, HBase 2.5.10, and the Kafka endpoints supplied by bigdata-platform. Compile-time success on a newer JDK does not establish cluster-runtime compatibility; upgrade the corresponding platform image and run the cluster deletion, persistence, and recall acceptance flows together.

For existing installations, stop jobs and back up checkpoints before replacing engines. Do not mix Spark 3/Scala 2.12 drivers with Spark 4/Scala 2.13 workers. A synthetic Spark 3.5.3 to 4.0.4 JSON-state restore was verified, but actual installation state still needs a copy-based rehearsal. Flink 1.x to 2.x savepoint compatibility is not guaranteed; preserve old savepoints and complete mutation history.

Testing

feature-core unit tests validate user/item aggregation, rolling windows, action counts, click-rate semantics, deduplication, and mutation-aware retraction. The shared golden fixture is executed by Python, feature-core, a real Flink keyed-state harness, and a Spark micro-batch test, including item-side expected rows, duplicates, invalid/future events, out-of-order insert/delete, and equal-time delete precedence. When changing a feature in rec-algorithm, update the shared contract and every fixture copy in the same change so online and offline definitions stay aligned.

mvn clean test is the local unit boundary. Redis, Kafka, HBase, Hive, checkpoint recovery, and DELETE tombstones are verified by the distribution-level cluster acceptance flow in example; do not treat one engine's unit suite as complete contract validation.

Refreshable serving windows

Feature snapshots include recentEventStats, aggregated per event second for the last 30 days. Each bucket contains count, value_sum, count:<action> and, for priced events, price_count:<action> / price_sum:<action>. These are derived from deduplicated, mutation-resolved state. Rank-engine uses them to refresh short-window counts/sums, windowed conversion rates and commerce price means without waiting for a new event. recentEventTimeCounts remains for old consumers.

This is an additive Redis snapshot contract; the Kafka mutation envelope and logical feature catalog are unchanged. Rebuild the streaming jobs and rank image together. Existing Redis snapshots lack the extra statistics until the producer emits them again; replay/backfill is necessary for inactive entities too. A rank restart alone does not reconstruct missing action or price history.