FlareDB I/O

FlareDbIO connects Apache Beam pipelines to FlareDB over Arrow Flight SQL. Use FlareDbIO.read() to run a query and read the result as Beam Rows, and FlareDbIO.write() to stream Rows into a FlareDB table.

Prerequisites

A running FlareDB instance. See the Quickstart.
The FlareDB I/O artifact in your project:

Gradle
Maven

build.gradle

dependencies {
    implementation 'com.flare-db:flareio-java:0.2.0'
    implementation 'com.flare-db:flaredb-runner:0.2.0'
}

pom.xml

<dependency>
  <groupId>com.flare-db</groupId>
  <artifactId>flareio-java</artifactId>
  <version>0.2.0</version>
</dependency>
<dependency>
  <groupId>com.flare-db</groupId>
  <artifactId>flaredb-runner</artifactId>
  <version>0.3.2</version>
</dependency>

Read from FlareDB

FlareDbIO.read() executes a SQL query on FlareDB and returns the results as a PCollection<Row>. The output schema is read from the query result.

ReadPipeline.java

PCollection<Row> rows =
    pipeline.apply(
        "ReadFromFlareDb",
        FlareDbIO.read()
            .fromQuery("SELECT id, name, score FROM flare.default.scores WHERE score > 90 ORDER BY id")
            .withDbUrl("grpc://localhost:8099"));

Method	Required	Description
`fromQuery(String)`	Yes	SQL query to run on FlareDB.
`withDbUrl(String)`	No	FlareDB URL. Defaults to `grpc://localhost:8099`.

Write to FlareDB

FlareDbIO.write() takes a PCollection<Row> and streams it into a FlareDB table. Schema is fixed per table, so every row written to a table must share the same Beam schema. If the destination table does not exist, FlareDbIO.write() creates it automatically using the Beam Row schema.

WritePipeline.java

rows.apply(
    "WriteToFlareDb",
    FlareDbIO.<Row>write()
        .to("flare.default.scores")
        .withDbUrl("grpc://localhost:8099"));

Method	Required	Description
`to(String)`	Yes	Destination table, for example `flare.default.scores`.
`withDbUrl(String)`	No	FlareDB URL. Defaults to `grpc://localhost:8099`.
`withBatchSize(int)`	No	Number of rows sent per Arrow record batch.

Rows are buffered within each Beam bundle and sent to FlareDB as Arrow record batches. Use withBatchSize() to control the maximum number of rows in each batch.

Table names

FlareDB addresses tables by fully qualified name: <catalog>.<database>.<table>. The catalog is flare and the database is default, so a scores table is flare.default.scores.

Complete example

WritePipeline.java

Pipeline pipeline = Pipeline.create(options);

PCollection<Row> rows =
    pipeline
        .apply("ReadCsv", TextIO.read().from("scores.csv"))
        .apply("ParseCsvToRows", ParDo.of(new CsvLineToRowFn()))
        .setRowSchema(SCORES_TABLE_SCHEMA);

rows.apply("WriteToFlareDb", FlareDbIO.<Row>write().to("flare.default.scores"));

pipeline.run();

Example pipelines

The repo includes runnable examples under flare-db/example:

flare-io-write — loads a CSV file into a FlareDB table
flare-io-read — reads that table back into a Beam pipeline as Rows

Together they cover the full lifecycle: write data into FlareDB, then query it either with SQL or a Beam pipeline.

Write example

flare-db/example/flare-io-write reads test-data/scores.csv with TextIO, parses each line into a Row using the schema id, name, city, age, score, and writes it to flare.default.scores with FlareDbIO.write().

PCollection<Row> rows =
    pipeline
        .apply("ReadCsv", TextIO.read().from(options.getCsvFile()))
        .apply("ParseCsvToRows", ParDo.of(new CsvLineToRowFn()))
        .setRowSchema(SCORES_TABLE_SCHEMA);

rows.apply(
    "WriteToFlareDb",
    FlareDbIO.<Row>write()
        .to(options.getTable())
        .withDbUrl(options.getDbUrl()));

Run it:

cd flare-db
./gradlew :flareio-write:shadowJar # build
./gradlew :flareio-write:run

This assumes FlareDB is running locally on 127.0.0.1:8099, and writes into flare.default.scores.

Read example

Once flare.default.scores has data, you can read it two ways. 1. Query it directly with flare-sql

# Open SQL shell
flare sql

Query

SELECT id, name, city, age, score
FROM default.default.scores
WHERE score > 90
ORDER BY id
LIMIT 5;

Expected result:

+----+---------+-----------+-----+-------+
| id | name    | city      | age | score |
+----+---------+-----------+-----+-------+
| 2  | Lakshmi | Vancouver | 26  | 94    |
| 6  | Priya   | Nairobi   | 30  | 91    |
| 50 | Asha    | Cape Town | 66  | 93    |
| 59 | Suresh  | Beijing   | 33  | 100   |
| 63 | Karthik | Hyderabad | 43  | 93    |
+----+---------+-----------+-----+-------+
5 rows returned (183 ms)

2. Read it into a pipeline with FlareDbIO.read() flare-db/example/flare-io-read runs the same query through FlareDbIO.read(), gets the result back as Beam Rows, and logs each row with a ParDo.

PCollection<Row> rows =
    pipeline.apply(
        "ReadFromFlareDb",
        FlareDbIO.read()
            .fromQuery("SELECT id, name, score FROM flare.default.scores WHERE score > 90 ORDER BY id")
            .withDbUrl(options.getDbUrl()));

Run it:

cd flare-db
./gradlew :flareio-read:shadowJar # build
./gradlew :flareio-read:run

Use flare sql for interactive analytics, and FlareDbIO.read() when you need to use the data as input to another Beam pipeline.

Intoduction

Guides

Prerequisites

Read from FlareDB

Write to FlareDB

Table names

Complete example

Example pipelines

Write example

Read example

​Prerequisites

​Read from FlareDB

​Write to FlareDB

​Table names

​Complete example

​Example pipelines

​Write example

​Read example

Prerequisites

Read from FlareDB

Write to FlareDB

Table names

Complete example

Example pipelines

Write example

Read example