---
title: lastBy Ingesters in a Core+ Worker
sidebar_label: lastBy Ingester
---
Deephaven Core+ workers can be used to ingest Deephaven binary logs and export a lastBy view over Barrage. Using a Core+ worker for a lastBy ingester with Barrage allows both Core+ and Legacy workers to subscribe to the resulting lastBy view. [Legacy lastBy Ingesters](../../legacy/importing-data/fast-lastby.md) export tables using a Java-serialization-based protocol, which cannot be used by a Core+ worker. Core+ Kafka ingesters may also generate [lastBy views from a Kafka stream](./coreplus-kafka.md#lastby-tables), which is not possible with the Legacy Kafka ingester.
This guide describes how to configure a lastBy [Data Import Server (DIS)](../dis.md) that receives [Deephaven binary logs](../binary-log-format.md) from the [tailer](../../sys-admin/configuration/data-tailer.md).
## Add an ImportState to the Schema
To configure a `lastBy` ingester, you must include an `ImportState` block in the schema. For a Core+ worker, the type of the import state must be `io.deephaven.enterprise.lastbystate.CoreLastByTableImportState`. In addition to specifying the class of the import state, you must provide column(s) which will be used for the `lastBy` aggregation.
If the schema is using a `LogFormat` block, the `ImportState` block is embedded within it. Each column which is aggregated is defined with its own `Column` block within the `ImportState`. For example, the following import state would generate a view for the `symbol` column.
```xml
```
If `symbol` and `exchange` are desired, then simply include a `Column` block for each of them within the context of the `ImportState` block.
```xml
```
Complete XML Schema Example (`symbol` and `exchange`) with `LogFormat` block
```xml
```
Add an ImportState to the Schema with `Listener` or `LoggerListener` block
> [!WARNING]
> `Listener` and `LoggerListener` blocks are deprecated, and may be removed in a future release. These blocks have been replaced with the `LogFormat` block.
To configure a lastBy ingester with a `Listener` or `LoggerListener` block, include an `ImportState` within the appropriate block. The `ImportState` must contain two attributes: `importStateType="io.deephaven.enterprise.lastbystate.CoreLastByTableImportState"`, and a `stateUpdateCall` which identifies the method invoked on each row received by the Data Import Server. The state update call is invoked on the import state class, and should specify the columns to use for your `lastBy` key. For example, the following import state would generate a `lastBy` view for the `symbol` column where the `newRow(...)` method of the `CoreLastByTableImportState` is used as the `stateUpdateCall` attribute.
```xml
```
If both `symbol` and `exchange` are desired, then simply use both columns as arguments to the state update call as follows:
```xml
```
Complete XML Schema Example (`symbol` and `exchange`) with `LoggerListener` block
```xml
```
## Create a Storage Directory
A lastBy DIS serves both lastBy and regular intraday queries. Therefore, it can serve as the sole storage location for the intraday data.
Intraday data is stored in a directory that must be exclusively owned by the DIS. Deephaven recommends the following location for the lastBy storage directory:
`/db/dataImportServers/[DIS_Name]/`
The lastBy storage directory must exist before the DIS Persistent Query can be started. When the script is run as a PQ on a merge server (preferred for production use), the `dbmerge` account will need read and write privileges. When the script is running in a regular query or console, the `dbquery` user on the server must have read and write access to this path.
## Configure Routing
The lastBy DIS must be added to the data routing configuration. See [Add a Data Import Server](../../sys-admin/configuration/yaml-add-dis.md) for instructions.
The following example adds a lastBy DIS named "lastByCore" with a claim for the `EqTrade` table in the `Example` namespace. The lastBy DIS uses dynamic endpoints, so port values are not supplied.
```bash
sudo -u irisadmin /usr/illumon/latest/bin/dhconfig dis add --name lastByCore --claim Example.EqTrade
```
This command adds a route with the following YAML:
```yaml
---
lastByCore:
endpoint:
serviceRegistry: registry
userIntradayDirectoryName: Users
throttleKbps: -1
claims:
- { namespace: Example, tableName: EqTrade }
storage: private
```
## Write an Ingestion Query
After configuring a schema, storage, and routing, the next step is to create an in-worker DIS for the desired table. The following script creates a DIS named "lastByCore". The routing YAML file uses "private" storage, so the second parameter to the `getDisByNameWithStorage` method specifies a pathname to a directory for the exclusive use of this in-worker DIS. The DIS automatically starts and listens on the Table Data and Tailer ports configured in routing. The above example uses a dynamic port registered with the service registry.
```groovy
import io.deephaven.enterprise.dataimportserver.DataImportServerTools
dis = DataImportServerTools.getDisByNameWithStorage("lastByCore", "/db/dataImportServers/lastByCore")
```
In addition to starting the DIS in the worker, we must create the lastBy view using the `io.deephaven.enterprise.lastbystate.LastByPartitionedTableFactory` (this must be done in the same worker that the DIS is running). There are three steps:
1. Create a `LastByPartitionedTableFactory` for the DIS.
2. Create a [PartitionedTable](/core/javadoc/io/deephaven/engine/table/PartitionedTable.html) for the namespace, table name, and column partition of interest. Each internal partition is represented as a constituent of the partitioned table.
3. Merge the partitioned table into a single view that contains the last row for a given key in each internal partition.
```groovy
import io.deephaven.enterprise.lastbystate.LastByPartitionedTableFactory
lbf = LastByPartitionedTableFactory.forDataImportServer(dis)
partitionedTable = lbf.createPartitionedTable("Example", "EqTrade", today())
lastTradeBySymbolExchange = partitionedTable.merge()
```
After the DIS is started, other queries can subscribe to the `lastTradeBySymbolExchange` variable using Barrage. As with other tables exported from a persistent query, you must grant view access to the DIS query and optionally [apply ACLs](../../sys-admin/permissions/permissions-overview.md). Note that the ACLs for the source table are not automatically applied to the exported result; the query writer must apply any desired ACLs.
## Retrieve the table from another query
The full history of the ingested data is retrieved as with any other table, in both Legacy and Core+ workers.
```groovy
trades = db.liveTable("Example", "EqTrade").where("Date=today()")
```
```python
trades = db.live_table("Example", "EqTrade").where(["Date=today()"])
```
The lastBy table can be retrieved from the in-worker DIS query using Barrage. In this example, the in-worker DIS is running in a PQ named `LastByQuery`:
```groovy
import io.deephaven.uri.ResolveTools
lastByTrades = ResolveTools.resolve("pq://LastByQuery/scope/lastTradeBySymbolExchange")
```
```python
from deephaven_enterprise import uri
last_by_trades = uri.resolve("pq://LastByQuery/scope/lastTradeBySymbolExchange")
```
## Related documentation
- [Data Import Server (DIS)](../dis.md)
- [Binary log format](../binary-log-format.md)
- [Tailer](../../sys-admin/configuration/data-tailer.md).