#stream #window #flink #kafka

rlink-kafka-connector

High performance Stream Processing Framework

30 releases

0.2.5 Mar 29, 2021
0.2.3 Mar 22, 2021
0.2.2-alpha.6 Feb 22, 2021
0.2.1-alpha.3 Jan 28, 2021
0.1.1 Dec 26, 2020

#3 in #flink

Download history 23/week @ 2020-12-22 6/week @ 2020-12-29 6/week @ 2021-01-05 42/week @ 2021-01-12 51/week @ 2021-01-19 101/week @ 2021-01-26 85/week @ 2021-02-02 25/week @ 2021-02-09 132/week @ 2021-02-16 16/week @ 2021-02-23 36/week @ 2021-03-02 83/week @ 2021-03-09 57/week @ 2021-03-16 84/week @ 2021-03-23 121/week @ 2021-03-30 34/week @ 2021-04-06

235 downloads per month
Used in cmdb-ip-mapping

MIT/Apache

555KB
14K SLoC

rlink-rs

Crates.io Released API docs MIT licensed License

High performance Stream Processing Framework. A new, faster, implementation of Apache Flink from scratch in Rust. pure memory, zero copy. single cluster in the production environment stable hundreds of millions per second window calculation.

Framework tested on Linux/MacOS/Windows, requires stable Rust.

Streaming Example

rlink = "0.2.0-alpha.5"
env.register_source(TestInputFormat::new(properties.clone()), 1)
    .assign_timestamps_and_watermarks(BoundedOutOfOrdernessTimestampExtractor::new(
        Duration::from_secs(1),
        SchemaBaseTimestampAssigner::new(model::index::timestamp, &FIELD_TYPE),
    ))
    .key_by(key_selector)
    .window(SlidingEventTimeWindows::new(
        Duration::from_secs(60),
        Duration::from_secs(20),
        None,
    ))
    .reduce(reduce_function, 2)
    .add_sink(PrintOutputFormat::new(output_schema_types.as_slice()));

Build

Build source

# debug
cargo build --color=always --all --all-targets
# release
cargo build --release --color=always --all --all-targets

Standalone Deploy

Config

standalone.yaml


---
# all job manager's addresses, one or more
application_manager_address:
  - "http://0.0.0.0:8770"
  - "http://0.0.0.0:8770"

metadata_storage:
  type: Memory

# bind ip
task_manager_bind_ip: 0.0.0.0
task_manager_work_dir: /data/rlink/application

task_managers

TaskManager list

10.1.2.1
10.1.2.2
10.1.2.3
10.1.2.4

Launch

Coordinator

./start_job_manager.sh

Worker

./start_task_manager.sh

Submit Application

On Standalone

## submit an application

# create job
curl http://x.x.x.x:8770/job/application \
  -X POST \
  -F "file=@/path/to/execute_file" \
  -v

# run job
curl http://x.x.x.x:8770/job/application/application-1591174445599 \
  -X POST \
  -H "Content-Type:application/json" \
  -d '{"batch_args":[{"cluster_mode":"Standalone", "manager_type":"Coordinator","num_task_managers":"15"}]}' \
  -v

# kill job
curl http://x.x.x.x:8770/job/application/application-1591174445599/shutdown \
  -X POST \
  -H "Content-Type:application/json"

On Yarn

update manager jar to hdfs

upload rlink-yarn-manager-0.2.0-alpha.5-jar-with-dependencies.jar to hdfs

eg: upload to hdfs://nn/path/to/rlink-yarn-manager-0.2.0-alpha.5-jar-with-dependencies.jar

update application to hdfs

upload your application executable file to hdfs.

eg: upload rlink-showcase to hdfs://nn/path/to/rlink-showcase

submit yarn job

submit yarn job with rlink-yarn-client-0.2.0-alpha.5.jar

hadoop jar rlink-yarn-client-0.2.0-alpha.5.jar rlink.yarn.client.Client \
  --applicationName rlink-showcase \
  --worker_process_path hdfs://nn/path/to/rlink-showcase \
  --java_manager_path hdfs://nn/path/to/rlink-yarn-manager-0.2.0-alpha.5-jar-with-dependencies.jar \
  --master_memory_mb 4096 \
  --master_v_cores 2 \
  --memory_mb 4096 \
  --v_cores 2 \
  --queue root.default \
  --cluster_mode YARN \
  --manager_type Coordinator \
  --num_task_managers 80

Dependencies

~32MB
~583K SLoC