flock-lab / flock Goto Github PK

It seems that for serverless stream processing, it's not a good solution to trigger the system for each new record. we’d better use Spark-like micro-batch processing instead of record-by-record processing. Only in this way can the system save money.

https://stackoverflow.com/questions/65491431/why-so-much-criticism-around-spark-streaming-micro-batch-when-using-kafka-as-so

https://github.com/DSLAM-UMD/ServerlessCQ/runs/1669688587

https://coveralls.io/

Since the physical plan is a nested structure, we need to recursively find all input nodes, and correctly map the data obtained from the parent lambda to each of them.

https://issues.apache.org/jira/browse/ARROW-11103

Automate account creation, and resource provisioning using AWS Service Catalog, AWS Organizations, and AWS Lambda

https://aws.amazon.com/cn/blogs/mt/automate-account-creation-and-resource-provisioning-using-aws-service-catalog-aws-organizations-and-aws-lambda/

After lambda functions with concurrency > 1 are dynamically executed, they are aggregated into the same subsequent lambda function through consistent hashing of the same seed in their payloads.

In the process of distributed computing, lambda functions with concurrency > 1 will send payloads to subsequent lambda function with concurrency = 1, in order to ensure that it can receive all payloads (not received by other functions of the same group with different names), We need to use the consistent hashing to deterministically select a subsequent function name from the same group. Consistent hashing with seed can address all potential problems.

Data Source

• S3, DynamoDB, AWS Aurora
• Streaming data in Lambda function

We can start with this PR in the future.

apache/arrow@77a46f2

/// Limit execution plan
#[derive(Debug, Serialize, Deserialize)]
pub struct GlobalLimitExec {
    /// Input execution plan
    input: Arc<dyn ExecutionPlan>,
    /// Maximum number of rows to return
    limit: usize,
    /// Number of threads to run parallel LocalLimitExec on
    concurrency: usize,
}

/// Sort execution plan
#[derive(Debug, Serialize, Deserialize)]
pub struct SortExec {
    /// Input schema
    input: Arc<dyn ExecutionPlan>,
    /// Sort expressions
    expr: Vec<PhysicalSortExpr>,
    /// Number of threads to execute input partitions on before combining into a single partition
    concurrency: usize,
}

we need to set concurrency value at runtime because the lambda instance is dynamic, changing depending on many factors.

Provisioned Concurrency is the end of cold starts, but at what costs for ServerlessCQ! [1]

1. https://aws.amazon.com/cn/blogs/aws/new-provisioned-concurrency-for-lambda-functions/

1. New York City Taxi & Limousine Commission Trip Record Data Link

The yellow and green taxi trip records include fields capturing pick-up and drop-off dates/times,
pick-up and drop-off locations, trip distances, itemized fares, rate types, payment types, and driver-reported passenger counts. The data is provided in CSV format.

US Accidents (3.5 million records) [Link]

This is a countrywide car accident dataset, which covers 49 states of the USA. The accident data are collected from February 2016 to June 2020, using two APIs that provide streaming traffic incident (or event) data. These APIs broadcast traffic data captured by a variety of entities, such as the US and state departments of transportation, law enforcement agencies, traffic cameras, and traffic sensors within the road-networks. Currently, there are about 3.5 million accident records in this dataset.

The customer is applying a continuous filter to only retain records of interest.

We need to modify some of the features of the open-source software called Arrow, such as serialization and deserialization of query's physical plan. We will regularly merge the latest official code into this forked version.

We will push all our changes to the branch scq:
https://github.com/DSLAM-UMD/arrow/tree/scq

In ServerlessCQ, if caching miss due to Lambda rescheduling, we can use the existing S3 Select feature to accelerate not only simple database operators like select and project, but also complex operators like join, group-by, and top-K.

Furthermore, starting today [1][2], Amazon S3 delivers strong read-after-write consistency automatically, which makes many of Starling's (SIGMOD'20) designs obsolete.

1. https://aws.amazon.com/s3/consistency/
2. https://news.ycombinator.com/item?id=25271791

We need to upgrade lambda crates, the current versions are incompatible to support the event source mapping for Kafka.

We should merge Arrow's official codebase at least once a month so that we can get the latest features.

#86

Users can issue continuous queries via either high-level language or CLI.

CLI features:

• Parse stream SQL queries, and generate multiple optimized physical plans through AWS state languages.
• Deploy AWS step functions automatically.
• Collect metrics from CloudWatch on Lambda functions (for adaptive optimization)

https://github.com/clap-rs/clap

Through the client's query engine, we can get the physical plan of the query statement. Before generating, compiling, and deploying lambda function code, we need to divide the entire plan into independent sub-plans, where each sub-plan corresponds to a lambda function.

The implementation logic of this part is:

input: physical plan tree

output: return topologically sorted sub plans

We have to find a way to serialize and deserialize some execution plans from datafusion that contain Vec<Arc<dyn traits>>.

/// Hash aggregate execution plan
#[derive(Debug)]
pub struct HashAggregateExec {
    mode: AggregateMode,
    group_expr: Vec<(Arc<dyn PhysicalExpr>, String)>,
    aggr_expr: Vec<Arc<dyn AggregateExpr>>,
    input: Arc<dyn ExecutionPlan>,
    schema: SchemaRef,
}

/// Execution plan for a projection
#[derive(Debug)]
pub struct ProjectionExec {
    /// The projection expressions stored as tuples of (expression, output column name)
    expr: Vec<(Arc<dyn PhysicalExpr>, String)>,
    /// The schema once the projection has been applied to the input
    schema: SchemaRef,
    /// The input plan
    input: Arc<dyn ExecutionPlan>,
}

• Build and upload generic lambda function to S3
• Deploy lambda functions and their environment context #137

If the input looks like the following JSON content:

{"a":1, "b":2.0, "c":false, "d":"4"}
{"a":-10, "b":-3.5, "c":true, "d":"4"}
{"a":2, "b":0.6, "c":false, "d":"text"}
{"a":1, "b":2.0, "c":false, "d":"4"}
{"a":7, "b":-3.5, "c":true, "d":"4"}
{"a":1, "b":0.6, "c":false, "d":"text"}
{"a":1, "b":2.0, "c":false, "d":"4"}
{"a":5, "b":-3.5, "c":true, "d":"4"}
{"a":1, "b":0.6, "c":false, "d":"text"}
{"a":1, "b":2.0, "c":false, "d":"4"}
{"a":1, "b":-3.5, "c":true, "d":"4"}
{"a":100000000000000, "b":0.6, "c":false, "d":"text"}

use arrow::datatypes::{DataType, Field, Schema};
use arrow::json;
use std::io::BufReader;
use std::sync::Arc;
use arrow::json::reader::infer_json_schema;

let inferred_schema = infer_json_schema(&mut BufReader::new(some_records), None).unwrap();
// if the window size is 1024
let mut json_reader = json::Reader::new(BufReader::new(whole_batch_records), Arc::new(inferred_schema), 1024, projection);  
let batch = json_reader.next().unwrap().unwrap(); //  batch is RecordBatch!

If we use Arrow's API directly, it seems that we don't need to convert JSON data into RecordBatch by ourselves.

The investigation is finished, I plan to write a new proposal with a feasible technical solution.

AWS Step Functions look like a fantastic solution to cope with the data flow stream model.

• Passing derived streams between state machines in AWS Step Functions

Question:

• What types of parameters does it support?
• Its size limit?
• How does it enable functions to transmit information?

crate:

1. Codegen Through handlebars, the lambda template is a more efficient way to generate lambda fuctions.

2. Compile code from code:

   • rustc_interface
   • example: rustc-example
   • cargo
   • cmd

Continuous query using SQL is one requirement of real-time stream processing [1].

This project has great potential to make another contribution to simplifying the serverless programming model via directly recasting stream queries into data flow models where each operator or node in the graph is a cloud function service managed by AWS Step Functions [3]. In this way, we can support graphs cyclic dataflows and iterations on streams.

Materialize does all of this by recasting SQL92 queries as dataflows. We can go further and directly convert SQL as FaaS dataflow. It's quite easy to parse SQL and support custom dialects [2].

References

1. The 8 Requirements of Real-Time Stream Processing. Michael Stonebraker, Uğur Çetintemel and Stan Zdonik
2. https://github.com/ballista-compute/sqlparser-rs
3. https://docs.aws.amazon.com/step-functions/latest/dg/welcome.html

People gradually began to realize this direction although it has not yet gone deep.

https://aws.amazon.com/blogs/compute/using-aws-lambda-for-streaming-analytics/

• parse time schedule and async trigger the repeated continuous query. @gangliao
• fetch window data (avoid duplicate reads) from data sources like Kinesis and Kafka in the 1st lambda function.
• read offline data (Parquet and CSV) from S3.
• window data partition for distributed data flow processing --- identify window number, sequence number, and total number, and so forth.
• record batch partition for multithreading processing inside the lambda function.
• understand Flink, Spark, and Presto (their related cloud services) for experimental evaluation. @ZejunLiu0
• collect useful open data set for both stream processing and batch processing
• how to handle lambda instance timeout (~15 min)
• compress/decompress and serialize/deserialize lambda context to cloud environment (4KB limit size) @gangliao
• Feed record batch to the leaf node of the subplan @gangliao

Based on the observation from #203

• ARROW-11310: [Rust] implement JSON writer
  apache/arrow@6c880b2#diff-8e3635c45cf23a781d2038b6a7aa9b28548e307b6351926d40012a4013f53078R38

an embedded key-value store to log queries and their cloud context into the disk.

https://github.com/spacejam/sled

In order to coalesce batches in the lambda function with concurrency = 1, shall we add a CoalesceBatchesExec to the head of the sub-plan?

Arrow Pull Request: umd-dslam/arrow#1

/// Physical expression of a scalar function
#[derive(Serialize, Deserialize)]
pub struct ScalarFunctionExpr {
    fun: ScalarFunctionImplementation,
    name: String,
    args: Vec<Arc<dyn PhysicalExpr>>,
    return_type: DataType,
}

/// Scalar function
pub type ScalarFunctionImplementation =
    Arc<dyn Fn(&[ArrayRef]) -> Result<ArrayRef> + Send + Sync>;

error[E0277]: the trait bound `dyn for<'r> Fn(&'r [Arc<(dyn arrow::array::Array + 'static)>]) -> std::result::Result<Arc<(dyn arrow::array::Array + 'static)>, DataFusionError> + Sync + std::marker::Send: physical_plan::_::_serde::Serialize` is not satisfied
   --> datafusion/src/physical_plan/functions.rs:391:5
    |
391 |     fun: ScalarFunctionImplementation,
    |     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ the trait `physical_plan::_::_serde::Serialize` is not implemented for `dyn for<'r> Fn(&'r [Arc<(dyn arrow::array::Array + 'static)>]) -> std::result::Result<Arc<(dyn arrow::array::Array + 'static)>, DataFusionError> + Sync + std::marker::Send`
    |
    = note: required because of the requirements on the impl of `physical_plan::_::_serde::Serialize` for `Arc<dyn for<'r> Fn(&'r [Arc<(dyn arrow::array::Array + 'static)>]) -> std::result::Result<Arc<(dyn arrow::array::Array + 'static)>, DataFusionError> + Sync + std::marker::Send>`

• Tumbling Window

• Stagger Window

• Sliding Window

• Session Window

• Support window function in SQL

Through the binary size optimization (https://github.com/johnthagen/min-sized-rust#), we are able to minimize the current binary size from 60MB to 665KB (with dependency libraries)!

-rw-r--r--.  1 gangliao gangliao 665K Oct 20 19:14 rust.zip

It's very easy to trigger its event on AWS.

In order to make performance comparisons more convenient, we MUST support Kafka.

https://github.com/fede1024/rust-rdkafka