Releases: ArroyoSystems/arroyo
v0.11.3
v0.11.3 is a patch release containing several fixes on top of 0.11.2. It is a drop-in replacement for clusters running any 0.11 patch release.
What's changed
v0.11.2
v0.11.2 is a patch release containing several fixes on top of 0.11.1. It is a drop-in replacement for clusters running 0.11.0 or 0.11.1.
What's changed
- Fix for checkpoint cleanup failure (#689)
- Use correct (relative) delta paths when writing to object stores (#693)
- Add support for IRSA authentication for S3 (#694)
Full Changelog: v0.11.1...v0.11.2
v0.11.1
0.11.1 is a patch release containing several bugfixes on top of 0.11.0. It is a drop-in replacement for clusters running 0.11.0.
What's changed
- Fix regression in updating equi-join optimization (#686)
- Add option to set raw_datums in avro schema editor (#685)
- Fix pending requests in preview that could stall out the webui (#684)
- Use an inner join to prevent controller from loading partially-constructed jobs (#681)
Full Changelog: v0.11.0...v0.11.1
v0.11.0
These release notes are also available on the Arroyo blog.
The Arroyo team is very excited to announce the release of Arroyo 0.11, which includes a number new features and improvements on top of 0.10, including pipeline clusters, a new, lightweight way to run streaming pipelines on serverless infrastructure, sqlite support for the control plane, a new configuration system, refinement of the create pipeline UX, and more.
In this release, we are thrilled to welcome two new contributors to the project:
- @benjamin-awd made their first contribution in #609
- @chokosabe made their first contribution in #652
Thanks to all of our contributors for this release:
Excited to try things out? Download a native package or run in Docker with
$ docker run -p 5115:5115 ghcr.io/arroyosystems/arroyo:0.11Pipeline clusters
Today Arroyo operates as a distributed cluster, with a controller node that manages a collection of pipelines, which themselves are scheduled on Arroyo nodes or Kubernetes.
Arroyo 0.11 introduces a new, lighter-weight way to run individual jobs, which we're calling pipeline clusters.
It works like this:
CREATE TABLE coinbase (
type TEXT,
price TEXT
) WITH (
connector = 'websocket',
endpoint = 'wss://ws-feed.exchange.coinbase.com',
subscription_message = '{
"type": "subscribe",
"product_ids": [
"BTC-USD"
],
"channels": ["ticker"]
}',
format = 'json'
);
select avg(cast(price as FLOAT)) from coinbase
group by hop(interval '1 second', interval '10 seconds');Here we've created a SQL file containing an Arroyo query and run it directly with the arroyo binary with the new arroyo run command. This starts up a completely self-contained minicluster that starts and manages the provided query. When the process is signalled to exit (via ctrl-c or SIGINT) the pipeline is stopped with a final checkpoint so it can be resumed later. If a sink isn't specified (as an INSERT INTO statement), the default STDOUT sink is used, which means you can consume pipeline outputs with UNIX pipes:
$ arroyo run query.sql > outputs.json
Pipeline clusters are great for running pipelines locally or in CI. But they also make it possible to run streaming pipelines in light-weight container runtimes like Amazon Fargate, Google Cloud Run, and Fly.io.
When running on a container runtime, state can be snapshotted and restored from an object store like S3 or GCS with the --state-dir flag. This means stateful, consistent, and reliable processing even on transient resources.
See the pipeline cluster docs for more details, and watch out for more tutorials and guides to running Arroyo on various serverless runtimes.
Sqlite control plane
In 0.10, we slimmed down the architecture of Arroyo, from a half-a-dozen services to three— one Arroyo service plus Postgres for configuration and metadata, and Prometheus to power the metrics features of the Web UI. In 0.11 we've finished this work:
- Sqlite is now supported as the storage engine for the control plane, powering the pipeline configuration
and API - We've replaced prometheus with a small, purpose-built metrics system to power the built-in metrics features
(with prometheus-compatible metrics still available for integration with external observability systems)
The entire Arroyo system can now run off a single, statically-linked, 150MB binary.
This greatly reduces the complexity and resource requirements of a minimal Arroyo deployment, which can now run comfortably in lightweight container hosting.
- Add Sqlite support for the control plane by @mwylde in #626
- Support metrics for web UI natively instead of using prometheus by @mwylde in #630
- Migrate arroyo-bin cmd to arroyo and consolidate docker images by @mwylde in #659
Configuration
Arroyo 0.11 ships with a completely new configuration system, replacing the ad-hoc environment variables used previously. This makes the system easier and more consistent to customize. We've also added comprehensive documentation on the various configuration options.
The new system is hierarchical; at the base is a
default config shipped with the engine. This can be overridden by Toml or Yaml files placed in the Users or system config directory (on Linux for example ~/.config/arroyo/config.toml), an arroyo.toml file in the current directory, then a config file passed on the command line with the new --config option. Finally, everything can be overridden by environment variables.
For example, on my Mac I have the following local config in my ~/Library/Application Support/arroyo/config.toml file
checkpoint-url = "/Users/mwylde/arroyo/checkpoints"
artifact-url = "/Users/mwylde/arroyo/artifacts"
[database]
type = "sqlite"On a per-run basis I can override this with environment variables, for example, to use Postgres instead
$ ARROYO__DATABASE__TYPE=postgres arroyo clusterSee the configuration docs for all of the options and full details on how the configuration system works.
Existing configuration environment variables (like SCHEDULER, DATABASE_HOST, etc.) will continue to be supported with a warning in 0.11, but will be removed in 0.12.
Create pipeline and preview improvements
We've reworked the UX for creating a pipeline and previewing it in the Web UI, focusing on performance and increased information density.
We've made the sidebars collapsible and integrated other UI elements to increase the amount of vertical and horizontal space for your queries and results.
The result table is now a high-performance virtual renderer, allowing us to increase the number of results shown to 10,000 (from 20 today), and the columns are now resizable, re-orderable, and sortable.
For power users there are few new options, including a checkbox to enable sinks in preview (by default they are disabled to avoid polluting external systems with test data) and set the parallelism on pipeline start.
- Redesign create pipeline UX and pipeline outputs for greater density and performance by @mwylde in #663
Raw bytes
Arroyo supports a number of formats (including JSON, Avro, and Parquet) but there will always be a long-tail of formats (potentially company-internal) that aren't built in.
Now it's possible to build your own decoders using UDFs and the new raw_bytes format.
When I started in streaming, the classic demo was analyzing Twitter data to find popular hashtags. Unfortunately, today Twitter API access costs thousands of dollars a month and is locked down. Fortunately there are now more open Twitter alternatives, like Bluesky which publishes all of its messages using the open AT protocol. And a website called Firesky has already done the hard work of aggregating these events into a websocket feed.
That would be great for Arroyo, except that its data is encoded in a binary format called cbor, a kind of binary JSON.
With raw_bytes and the serde-json crate this is no longer an issue!
We'll start with a simple UDF that converts cbor to JSON:
/*
[dependencies]
serde_cbor = "0.11"
serde_json = "1"
serde = {version = "1", features = ["derive"]}
serde-transcode = "1"
*/
use arroyo_udf_plugin::udf;
#[udf]
fn cbor_to_json(data: &[u8]) -> Option<String> {
let mut deserializer = serde_cbor::Deserializer::from_slice(data);
let mut buf = std::io::BufWriter::new(Vec::new());
let mut serializer = serde_json::Serializer::new(&mut buf);
serde_transcode::transcode(&mut deserializer, &mut serializer).ok()?;
let bytes = buf.into_inner().unwrap();
Some(String::from_utf8(bytes).ok()?)
}(we could have also processed directly on the cbor data, but the tools for dealing with JSON are a bit more convenient).
Then, we can use this in a query to count the hashtags:
create table firesky (
value BYTEA
) with (
connector = 'websocket',
endpoint = 'wss://firesky.tv/ws/app',
format = 'raw_bytes'
);
create view tags as (
s...v0.10.3
0.10.3 is a patch release containing a couple of fixes on top of 0.10.2. It is a drop-in replacement for clusters running any patch release of 0.10.
What's changed
- Add table for kafka exactly-once state by @mwylde in #655
- Add option to deploy kubernetes programs as configmaps rather than env vars by @mwylde in #654
Full Changelog: v0.10.2...v0.10.3
Note that to take advantage of the fix for the issue in #654, the helm chart must be upgraded to 0.10.3 as well (the existing helm chart will work with the new image, however, the fix is disabled as it requires additional k8s permissions).
Contributors
Assets 5
v0.10.2
0.10.2 is a patch release containing several fixes on top of 0.10.1. It is a drop-in replacement for clusters running 0.10.0 or 0.10.1.
What's Changed
- Statically compiled lzma to prevent dynamic linking on macos by @mwylde in #614
- Fix regression in raw_json schema conversion by @mwylde in #616
- flush kafka on close by @jacksonrnewhouse in #619
- Run apt-get update before trying to install clang by @mwylde in #622
Full Changelog: v0.10.1...v0.10.2
Contributors
Assets 5
v0.10.1
0.10.1 is a patch release containing several fixes on top of 0.10.0. It is a drop-in replacement for clusters running 0.10.0.
- Fix regression in bad_data handling by @mwylde in #606
- Strip file:// off of the final_dir for local sinks by @jacksonrnewhouse in #607
- Add pkg-config to Dockerfile by @benjamin-awd in #609
- Rework async_plan and extend stack size in debug mode by @jacksonrnewhouse in #613
- Support writing non-updating queries to debezium sinks by @jacksonrnewhouse in #612
Full Changelog: v0.10.0...v0.10.1
Contributors
Assets 4
v0.10.0
These release notes are also available on the Arroyo blog.
Arroyo 0.10.0 is a major release of the Arroyo stream processing engine, featuring an entirely new SQL engine that's >3x faster and ships as a single binary. Plus NATS and MQTT connectors, more SQL features, and more.
NATS
NATS is a messaging and queueing system designed for simplicity and performance. Arroyo 0.10 adds sources and sinks for Core NATS and NATS Jetstream, which layers on persistence and delivery guarantees.
For example, to consume from a NATS Jetstream subject, we can use this DDL:
CREATE TABLE logs (
id BIGINT NOT NULL,
time TIMESTAMP NOT NULL,
host TEXT NOT NULL,
level TEXT NOT NULL,
message TEXT NOT NULL
) with (
type = 'source',
connector = 'nats',
servers = 'localhost:4222',
'nats.subject' = 'logs',
'auth.type' = 'credentials',
'auth.username' = '{{ NATS_USER }}',
'auth.password' = '{{ NATS_PASSWORD }}',
format = 'json'
);See the connector docs for more details.
Thanks to Quentin Gaborit for this incredible contribution!
MQTT
MQTT is a lightweight messaging protocol widely used with low-power devices and in the Internet of Things (IoT) space. Arroyo 0.10 now ships with an MQTT source and sink to consume and produce from MQTT brokers.
For example, you can create an MQTT source with the following SQL
CREATE TABLE devices (
device_id BIGINT NOT NULL,
time TIMESTAMP NOT NULL,
lat FLOAT NOT NULL,
lng FLOAT NOT NULL,
metadata JSON
) with (
connector = 'mqtt',
url = 'tcp://localhost:1883',
type = 'source',
topic = 'events',
format = 'json'
);See the connector docs for more details.
Thanks to community member Giovanny Gutiérrez (@bakjos) for this amazing contribution!
$ arroyo
With the new architecture in 0.10, the entire Arroyo system now builds as a single binary
$ arroyo
Usage: arroyo <COMMAND>
Commands:
api Starts an Arroyo API server
controller Starts an Arroyo Controller
cluster Starts a complete Arroyo cluster
worker Starts an Arroyo worker
compiler Starts an Arroyo compiler
node Starts an Arroyo node server
migrate Runs database migrations on the configure Postgres database
help Print this message or the help of the given subcommand(s)
Options:
-h, --help Print help
-V, --version Print versionRunning a complete local cluster is as easy as
$ arroyo cluster
INFO arroyo_controller: Using process scheduler
INFO arroyo_server_common: Starting cluster admin server on 0.0.0.0:8001
INFO arroyo_controller: Starting arroyo-controller on 0.0.0.0:9190
INFO arroyo_api: Starting API server on 0.0.0.0:8000
INFO arroyo_compiler_service: Starting compiler service at 0.0.0.0:9000Today we are providing pre-compiled binaries on the release page for Linux and MacOS, or you can build your own by following the dev instructions.
Note that Arroyo currently requires a running Postgres database for configuration, but in the next release we will be adding the option of using sqlite. Prometheus is also used to power the pipeline metrics in the Web UI if available.
Performance
Arroyo 0.10 is significantly faster than 0.9.
In fact, it's so fast that for many benchmarks we've run on EC2 instances, it can saturate the network before using all the available CPU. We'll be following up with rigorous benchmarks against other systems, but in early tests it significantly beats other streaming systems in throughput.
How has it gotten so much faster? This is deserving of an entire blog post, but the short version is that Arroyo now operates internally on columnar data using carefully tuned vector kernels, thanks to the hard work of the Apache Arrow and DataFusion communities.
Columnar data is now the standard for OLAP (analytics-oriented) query engines like ClickHouse, Pinot, and Presto. There are a few reasons for this:
- By storing all values in a column together, you can achieve better compression ratios and make better use of CPU cache
- Only the columns actually referenced in a query need to be read, reducing disk and network IO
- Columnar processing aligns well with the vector capabilities in modern CPUs, providing 3x or more speedups
However, row-oriented data remains the standard for streaming engines. There are some inherent tradeoffs between latency (how quickly an event can traverse through the pipeline) and throughput (how many events can be processed with a given amount of CPU). By batching data we can get higher throughput at the expense of latency. And columnar representations require that we batch a number of events together before we see performance improvements (in fact, with a small number of rows columnar processing will be much slower due to fixed overhead).
But there's a strong argument for moving to columnar representations for streaming: at any given batch size, the higher the throughput the less time we must wait to fill that batch. For example, if we want at least 100 records in our batch to overcome fixed costs, the amount of time we need to wait to receive 100 records will depend on our throughput:
- At 10 events/second, it takes 1 second
- At 1,000 — 0.01 seconds (100ms)
- At 1,000,000 — 0.0001 (0.1ms)
Or looking at it from a fixed latency perspective (say, waiting at most 10ms):
- At 10 events/second, our batch size is 1
- At 1,000 — 100
- At 1,000,000 — 100,000
In Arroyo 0.10, we perform batching at the source and via aggregating operators like windows and joins. The source batching behavior can be configured via two environment variables:
BATCH_SIZEwhich controls the maximum batch sizeBATCH_LINGER_MSwhich controls the maximum amount of time to wait (in milliseconds) before emitting a batch
By configuring these values, users can choose their own tradeoff between latency and throughput.
SQL Improvements
Functions
We've added support for over 200 new scalar, aggregate, and window SQL functions. These unlock many powerful new capabilities that previously would have been inexpressible or would have required writing a UDF.
There are too many new functions to highlight them all, but some exciting additions include:
- Nearly 200 scalar functions, covering math, conditions, strings, regex, JSON, times, and more
- Statistical and approximate aggregate functions like
approx_percentile_contandapprox_distinct,which uses HyperLogLog to produce very memory-efficient estimate distinct counts - Many new SQL window functions including rank, percent_rank, lag, and lead
Check out the Scalar, Aggregate,
and Window function docs for the complete list.
Arrays
Previously, Arroyo supported a limited set of operations on arrays, but only within certain contexts. Now we have complete array support, including a comprehensive set of array function, support for indexing via square brackets (like a[1]), support for serializing arrays as JSON and Avro, the ability to convert aggregates into arrays (via array_agg), and the ability to unroll arrays into separate rows via unnest.
For example, we can write unique IPs over a tumbling window as an array to Kafka with a query like this:
CREATE TABLE sink (
values TEXT[]
) with (
connector = 'confluent',
connection_profile = 'confluent-working',
format = 'avro',
'avro.confluent_schema_registry' = 'true',
topic = 'list_output',
type = 'sink'
);
INSERT INTO sink
SELECT array_agg(DISTINCT ip) FROM logs
GROUP BY tumble(interval '10 seconds');Source Fusion
Arroyo 0.10 includes new optimizations that reduce the amount of data read from a single source when it's used across multiple queries. Now, each source will only be read once rather than once per subquery.
This same optimization also applies to view, so compute can be reused across outputs.
Upgrading to 0.10
There are some backwards-incompatible changes in 0.10's SQL. Some pipelines written for 0.9 will need to be updated as described here.
Virtual fields
The syntax for virtual fields has changed to more closely match Postgres: we now require the STORED keyword. Previously:
event_time TIMESTAMP GENERATED ALWAYS AS (CAST(date as TIMESTAMP))Now:
event_time TIMESTAMP GENERATED ALWAYS AS (CAST(date as TIMESTAMP)) STOREDUDFs
UDF definitions have changed slightly; now the UDF function must be annotated with an attribute macro #[arroyo_udf_plugin::udf], for example:
use arroyo_udf_plugin::udf;
#[udf]
fn plus_one(x: i64) -> i64 {
x + 1
}This change means we now support defining other ...
v0.9.1
0.9.1 is a patch release containing several fixes on top of 0.9.0. It is a drop-in replacement for clusters running 0.9.0.
- Upgrade framer-motion to fix UDF popover by @jbeisen in #499
- Fix cases where kafka client_config setting was not used by @mwylde in #520
Full Changelog: v0.9.0...v0.9.1
Contributors
Assets 2
V0.9.0
These release notes are also available on the Arroyo blog.
Arroyo 0.9.0 introduces async UDFs, which allow users to use databases, services, and models from within their pipelines. It also brings support for joining update tables, more control over bad data handling, a redesigned connection profile editor, and more.
For this release, we are thrilled to welcome two new contributors to the Arroyo project:
- @tqwewe made their first contribution in #456
- @breezykermo made their first contribution in #476
Thanks to all our contributors for this release:
Async UDFs
User-defined functions (UDFs) and user-defined aggregate functions (UDAFs) allow you to extend Arroyo with custom logic. New in Arroyo 0.9 is support for what we call async UDFs.
Existing (sync) UDFs are expected to implement simple, computational logic. Common use cases include parsing custom formats, implementing functions that are not natively supported, or implementing custom business logic that would be difficult to express in SQL. Because they are run synchronously with processing, if they take too long to run they can block the entire pipeline.
This isn't a hypothetical concern. A UDF that takes 10ms to run will be limited to processing just 100 events per second per subtask!
And there are many use cases where you want to run logic that is not a simple, stateless computation. You may need to do point lookups in a database to enrich events, make an API call to another service, or even perform inference against an AI model.
Async UDFs allow you to do all of these things, without blocking the pipeline. Async UDFs are defined as a Rust async fn, supporting non-blocking IO. Then within the Arroyo runtime, many instances of the UDF can be run in parallel, with a configurable concurrency limit.
What does this look like? Let's take an example of a UDF that enriches web events
with GeoIP data by calling a GeoIP service. First, we define the UDF:
/*
[dependencies]
reqwest = { version = "0.11.23", features = ["json"] }
serde_json = "1"
[udfs]
async_max_concurrency = 1000
*/
pub async fn get_city(ip: String) -> Option<String> {
let body: serde_json::Value =
reqwest::get(
format!("http://geoip-service:8000/{ip}"))
.await
.ok()?
.json()
.await
.ok()?;
body.pointer("/names/en").and_then(|t|
t.as_str()
).map(|t| t.to_string())
}Then we can use this UDF in a query, for example this one that finds the most common cities in the last 15 minutes:
create view cities as
select get_city(logs.ip) as city
from logs;
SELECT * FROM (
SELECT *, ROW_NUMBER() OVER (
PARTITION BY window
ORDER BY count DESC) as row_num
FROM (SELECT count(*) as count,
city,
hop(interval '5 seconds', interval '15 minutes') as window
FROM cities
WHERE city IS NOT NULL
group by city, window)) WHERE row_num <= 5;Async UDFs support several configuration options that control their behavior, including the maximum number of concurrent requests, timeouts, and whether event order should be preserved.
They also support defining a Context struct which is passed in to each invocation of the UDF. This allows you to do setup for expensive operations (like setting up a Database connection pool) just once, and share the result with all invocations.
For example, we can create a client for Postgres like this:
pub struct Context {
client: RwLock<Option<Client>>,
}
impl Context {
pub fn new() -> Self {
Self {
client: RwLock::new(None),
}
}
}
#[async_trait]
impl arroyo_types::UdfContext for Context {
async fn init(&self) {
let conn_str = "host=localhost user=dbuser password=dbpassword dbname=my_db";
let (client, connection) = tokio_postgres::connect(conn_str, NoTls).await.unwrap();
let mut c = self.client.write().await;
*c = Some(client);
tokio::spawn(async move {
if let Err(e) = connection.await {
println!("connection error: {}", e);
}
});
}
}See the docs for full details.
- Add Async UDFs by @jbeisen in #483
- Enable user-defined context structs for async UDFs by
@jbeisen in #487
Joining Update tables
Arroyo has two semantics for streaming SQL, one based on event-time watermarks and another that we call "Update Tables," which allow for incremental computation of most analytical SQL constructs. However, previously there was a restriction on joins, which were only supported for the watermark semantic.
Now that restriction is gone, and update tables can be joined. Hurray!
For example, this query will count the number of page views by user, and when a transaction comes in joins that accumulated count:
CREATE VIEW page_view_counts as
SELECT count(*) as count, user_id as user_id
FROM page_views
GROUP BY user_id;
SELECT T.user_id, amount, P.count as page_views
FROM transactions T
LEFT JOIN page_view_counts P on T.user_id = P.user_id;- Support joining updating data by @jbeisen in #420
- Just retract the final row of an outer join, rather than (None, None) by
@jacksonrnewhouse in #490
Bad Data handling
It happens to the best of us. You've carefully built out your data architecture, perform rigorous code reviews, and diligently monitor your rollouts.
And yet it happens—somehow, invalid data got into your Kafka topic. Now in Arroyo 0.9, you can now configure the source to drop bad data instead of causing a processing failure.
This behavior can be configured via the Web UI or with the bad_data option when creating a source in SQL. Two options are currently available:
-
fail(default, and behavior in Arroyo 0.8) causes the pipeline to restart when bad data is encountered -
dropcauses the data to be dropped, while logging an error message to the console and incrementing thearroyo_worker_deserialization_errorsmetric -
Add control for deserialization error behavior by @jbeisen in #443
-
Add bad data option to create connection form by @jbeisen in #452
Environment variable substitution
Some connectors need to be configured with authentication details or other secret data. Today in Arroyo those secrets are stored in the configuration database (postgres), which is not very secure.
Now in Arroyo 0.9, we're introducing environment-variable substitution for sensitive configuration fields:
This feature allows you to use double curly braces ({{ }}) to reference environment variables, which will get substituted in at run time. For example, if you have an environment variable called WEBHOOK_SECRET you can now include this as a header in a webhook sink like
Authentication: Basic {{ WEBHOOK_SECRET }}
Env variable substitution can be used for both connections created via the Web UI and those created directly in SQL, like:
CREATE TABLE slack (
value TEXT
) WITH (
connector = 'webhook',
endpoint = 'https://hooks.slack.com/services/{{ SLACK_KEY }}',
method = 'POST',
headers = 'Content-Type:application/json',
format = 'json'
);Confluent Cloud connector
Confluent is the company founded by the creators of Apache Kafka. They provide a cloud-native distribution of Kafka and serverless Kafka platform. While Arroyo has always supported reading and writing from Kafka, with Arroyo 0.9 we're making it even easier to integrate with your Confluent Cloud topics with the new Confluent Cloud connector.
See the complete integration guide to get started.
Connection Profile UI
Connection Profiles encapsulate common configuration that is shared across multiple connection tables. For example, a single Kafka cluster may have many topics that you would like to consumer or produce to from Arroyo.
We've upgraded the process of creating and managing these profiles. It's now possible to view and delete existing connection profiles, and the whole UI has gotten a little spiffier.
We're also now validating that the profile is good (like ensuring we can talk to Kafka with the provided address and credentials), so you don't discover you mistyped something at the end of the connection creation process.
- Redesign cluster profile UI and add v...
