Documentation

InfluxDB OSS

v2

Join data with Flux

Use the Flux join package to join two data sets based on common values using the following join methods:

Inner join

Left outer join

Right outer join

Full outer join

The join package lets you join data from different data sources such as InfluxDB, SQL database, CSV, and others.

Use join functions to join your data

Inner join Left join Right join Full outer join Join on time

  1. Import the join package.

  2. Define the left and right data streams to join:

    • Each stream must have one or more columns with common values. Column labels do not need to match, but column values do.
    • Each stream should have identical group keys.

    For more information, see join data requirements.

  3. Use join.inner() to join the two streams together. Provide the following required parameters:

    • left: Stream of data representing the left side of the join.
    • right: Stream of data representing the right side of the join.
    • on: Join predicate. For example: (l, r) => l.column == r.column.
    • as: Join output function that returns a record with values from each input stream. For example: (l, r) => ({l with column1: r.column1, column2: r.column2}).
import "join"
import "sql"

left =
    from(bucket: "example-bucket-1")
        |> range(start: "-1h")
        |> filter(fn: (r) => r._measurement == "example-measurement")
        |> filter(fn: (r) => r._field == "example-field")

right =
    sql.from(
        driverName: "postgres",
        dataSourceName: "postgresql://username:password@localhost:5432",
        query: "SELECT * FROM example_table",
    )

join.inner(
    left: left,
    right: right,
    on: (l, r) => l.column == r.column,
    as: (l, r) => ({l with name: r.name, location: r.location}),
)

For more information and detailed examples, see Perform an inner join in the Flux documentation.

  1. Import the join package.

  2. Define the left and right data streams to join:

    • Each stream must have one or more columns with common values. Column labels do not need to match, but column values do.
    • Each stream should have identical group keys.

    For more information, see join data requirements.

  3. Use join.left() to join the two streams together. Provide the following required parameters:

    • left: Stream of data representing the left side of the join.
    • right: Stream of data representing the right side of the join.
    • on: Join predicate. For example: (l, r) => l.column == r.column.
    • as: Join output function that returns a record with values from each input stream. For example: (l, r) => ({l with column1: r.column1, column2: r.column2}).
import "join"
import "sql"

left =
    from(bucket: "example-bucket-1")
        |> range(start: "-1h")
        |> filter(fn: (r) => r._measurement == "example-measurement")
        |> filter(fn: (r) => r._field == "example-field")

right =
    sql.from(
        driverName: "postgres",
        dataSourceName: "postgresql://username:password@localhost:5432",
        query: "SELECT * FROM example_table",
    )

join.left(
    left: left,
    right: right,
    on: (l, r) => l.column == r.column,
    as: (l, r) => ({l with name: r.name, location: r.location}),
)

For more information and detailed examples, see Perform a left outer join in the Flux documentation.

  1. Import the join package.

  2. Define the left and right data streams to join:

    • Each stream must have one or more columns with common values. Column labels do not need to match, but column values do.
    • Each stream should have identical group keys.

    For more information, see join data requirements.

  3. Use join.right() to join the two streams together. Provide the following required parameters:

    • left: Stream of data representing the left side of the join.
    • right: Stream of data representing the right side of the join.
    • on: Join predicate. For example: (l, r) => l.column == r.column.
    • as: Join output function that returns a record with values from each input stream. For example: (l, r) => ({l with column1: r.column1, column2: r.column2}).
import "join"
import "sql"

left =
    from(bucket: "example-bucket-1")
        |> range(start: "-1h")
        |> filter(fn: (r) => r._measurement == "example-measurement")
        |> filter(fn: (r) => r._field == "example-field")

right =
    sql.from(
        driverName: "postgres",
        dataSourceName: "postgresql://username:password@localhost:5432",
        query: "SELECT * FROM example_table",
    )

join.right(
    left: left,
    right: right,
    on: (l, r) => l.column == r.column,
    as: (l, r) => ({l with name: r.name, location: r.location}),
)

For more information and detailed examples, see Perform a right outer join in the Flux documentation.

  1. Import the join package.

  2. Define the left and right data streams to join:

    • Each stream must have one or more columns with common values. Column labels do not need to match, but column values do.
    • Each stream should have identical group keys.

    For more information, see join data requirements.

  3. Use join.full() to join the two streams together. Provide the following required parameters:

    • left: Stream of data representing the left side of the join.
    • right: Stream of data representing the right side of the join.
    • on: Join predicate. For example: (l, r) => l.column == r.column.
    • as: Join output function that returns a record with values from each input stream. For example: (l, r) => ({l with column1: r.column1, column2: r.column2}).

Full outer joins must account for non-group-key columns in both l and r records being null. Use conditional logic to check which record contains non-null values for columns not in the group key. For more information, see Account for missing, non-group-key values.

import "join"
import "sql"

left =
    from(bucket: "example-bucket-1")
        |> range(start: "-1h")
        |> filter(fn: (r) => r._measurement == "example-measurement")
        |> filter(fn: (r) => r._field == "example-field")

right =
    sql.from(
        driverName: "postgres",
        dataSourceName: "postgresql://username:password@localhost:5432",
        query: "SELECT * FROM example_table",
    )

join.full(
    left: left,
    right: right,
    on: (l, r) => l.id== r.id,
    as: (l, r) => {
        id = if exists l.id then l.id else r.id
        
        return {name: l.name, location: r.location, id: id}
    },
)

For more information and detailed examples, see Perform a full outer join in the Flux documentation.

  1. Import the join package.

  2. Define the left and right data streams to join:

    • Each stream must also have a _time column.
    • Each stream must have one or more columns with common values. Column labels do not need to match, but column values do.
    • Each stream should have identical group keys.

    For more information, see join data requirements.

  3. Use join.time() to join the two streams together based on time values. Provide the following parameters:

    • left: (Required) Stream of data representing the left side of the join.
    • right: (Required) Stream of data representing the right side of the join.
    • as: (Required) Join output function that returns a record with values from each input stream. For example: (l, r) => ({r with column1: l.column1, column2: l.column2}).
    • method: Join method to use. Default is inner.
import "join"
import "sql"

left =
    from(bucket: "example-bucket-1")
        |> range(start: "-1h")
        |> filter(fn: (r) => r._measurement == "example-m1")
        |> filter(fn: (r) => r._field == "example-f1")

right =
    from(bucket: "example-bucket-2")
        |> range(start: "-1h")
        |> filter(fn: (r) => r._measurement == "example-m2")
        |> filter(fn: (r) => r._field == "example-f2")

join.time(method: "left", left: left, right: right, as: (l, r) => ({l with f2: r._value}))

For more information and detailed examples, see Join on time in the Flux documentation.

When to use union and pivot instead of join functions

We recommend using the join package to join streams that have mostly different schemas or that come from two separate data sources. If you’re joining two datasets queried from InfluxDB, using union() and pivot() to combine the data will likely be more performant.

For example, if you need to query fields from different InfluxDB buckets and align field values in each row based on time:

f1 =
    from(bucket: "example-bucket-1")
        |> range(start: "-1h")
        |> filter(fn: (r) => r._field == "f1")
        |> drop(columns: "_measurement")

f2 =
    from(bucket: "example-bucket-2")
        |> range(start: "-1h")
        |> filter(fn: (r) => r._field == "f2")
        |> drop(columns: "_measurement")

union(tables: [f1, f2])
    |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")

View example input and output data

join flux

Was this page helpful?

Thank you for your feedback!

© 2024 InfluxData, Inc.

What is your InfluxDB OSS URL?

Customize your InfluxDB OSS URL and we’ll update code examples for you.

Default
Custom

For more information, see InfluxDB OSS URLs.

Thank you for your feedback!

Let us know what we can do better:

Thank you!

No thanks

The future of Flux

Flux is going into maintenance mode. You can continue using it as you currently are without any changes to your code.

Read more

InfluxDB v3 enhancements and InfluxDB Clustered is now generally available

New capabilities, including faster query performance and management tooling advance the InfluxDB v3 product line. InfluxDB Clustered is now generally available.

InfluxDB v3 performance and features

The InfluxDB v3 product line has seen significant enhancements in query performance and has made new management tooling available. These enhancements include an operational dashboard to monitor the health of your InfluxDB cluster, single sign-on (SSO) support in InfluxDB Cloud Dedicated, and new management APIs for tokens and databases.

Learn about the new v3 enhancements

InfluxDB Clustered general availability

InfluxDB Clustered is now generally available and gives you the power of InfluxDB v3 in your self-managed stack.

Talk to us about InfluxDB Clustered