Working with Fabric Eventhouses (real-time data)

An Eventhouse stores event, log, and time-series data in KQL databases. In this guide, ingestion means adding rows to a KQL table. Start with the discovered database’s $write_table() method (fabric_kql_write_table()), which accepts an R data frame and manages staging, status checks, and cleanup for you.

After the basic write, this guide shows how to monitor ingestion, use an existing storage file, scale up with Arrow, and export a large query result. The lower-level ingestion routes used by these workflows are currently in preview.

Find a KQL database

Start with a discovered KQL database. The result is a read-only FabricKqlDatabase R6 object. Its fields contain both the query and ingestion URIs, and its methods reuse that context and the discovery credential:

library(fabricQueryR)

database <- fabric_kql_databases("Telemetry workspace")[[1]]
database$ingestion_service_uri

Write a small data frame in one call

Supply a destination table and an ordinary data frame or tibble. Use a new test table while learning:

written <- database$write_table(
  table = "Events",
  data = data.frame(
    id = 1:3,
    category = c("A", "B", "A"),
    amount = c(10.5, 20, 30.5)
  ),
  create_if_missing = TRUE,
  ingest_if_not_exists = "r-events-2026-08-14"
)

written$status$state
written$rows
written$staging_retained

The function writes temporary Parquet data, uploads it to Fabric, queues the ingestion, waits for a final status, and deletes staging only after confirmed success. create_if_missing = TRUE creates a basic table from the R object’s columns when needed.

For an existing table, the source names and types must match. Supply a predefined Parquet mapping or explicit column_types when inference is not appropriate. See ?fabric_kql_write_table for supported type mappings and staging recovery options.

Queue a tracked batch

Use the lower-level $ingest() method (fabric_kql_ingest()) when the source file already exists in OneLake or supported blob storage. It does not upload a local file or serialize an R object, and the destination table must already exist.

The mapping argument is optional. When it is omitted, Kusto derives an identity mapping from the existing table schema: ordered text formats such as CSV map source columns by position, while JSON, Parquet, Avro, ORC, and W3CLOGFILE map fields to case-sensitive table-column names. Use a validated named mapping when source order or names differ, or when ingestion-time transformations are required. Supply the complete source path and its format:

source <- paste0(
  "https://onelake.dfs.fabric.microsoft.com/",
  "<workspace-id>/<lakehouse-id>",
  "/Files/events/2026-08-14.csv;impersonate"
)

The signed-in identity must be able to read the file. Do not print or log a source string that contains credentials. Source IDs are generated when omitted and remain available on the returned handle. This example uses identity mapping unless FABRIC_KQL_INGESTION_MAPPING names a predefined CSV mapping:

mapping <- Sys.getenv("FABRIC_KQL_INGESTION_MAPPING", unset = "")

ingestion <- database$ingest(
  table = "Events",
  sources = source,
  format = "csv",
  mapping = if (nzchar(mapping)) mapping else NULL,
  ignore_first_record = TRUE,
  tags = "source:daily-export",
  ingest_if_not_exists = "events-2026-08-14"
)

ingestion$id
ingestion$sources$source_id

Use a stable ingest_if_not_exists key when the same source file may be submitted again. Idempotency keys require one source per call; submit multiple files separately with a distinct stable key for each file. Queued ingestion is an advanced, at-least-once workflow: after an uncertain network result, inspect the tracked operation and target table before submitting the source again. See ?fabric_kql_ingest for batching, source deletion, and storage-authentication options.

Wait and inspect the outcome

One status snapshot with $ingestion_status() (fabric_kql_ingestion_status()) is useful for a scheduler that persists operation IDs:

snapshot <- database$ingestion_status(ingestion)
snapshot$state
snapshot$counts

For an interactive or single-process batch, $ingestion_wait() calls fabric_kql_ingestion_status(..., wait = TRUE) with a client-side deadline:

result <- database$ingestion_wait(
  ingestion,
  timeout = 900,
  poll_interval = 2
)

result$state
result$details

The deadline stops only the R waiter; it does not cancel ingestion in Fabric. By default, a failed or partially successful batch raises an R condition. To inspect every terminal state as data instead, call $ingestion_wait() (fabric_kql_ingestion_status(..., wait = TRUE)) with error_on_failure = FALSE:

result <- database$ingestion_wait(
  ingestion,
  error_on_failure = FALSE
)

failed <- subset(
  result$details,
  status %in% c("Failed", "Canceled")
)
failed[c("source_id", "error_code", "failure_status", "message")]

Query the destination after success with $query() (fabric_kql_query()):

loaded <- database$query(
  query = "Events | where ingestion_time() > ago(1h) | take 100"
)

The caller needs permission to ingest into the destination table and read the source file. Keep a manually staged source until the tracked result is final and verified; $write_table() (fabric_kql_write_table()) manages this retention rule for you.

Scale up with Arrow

For data larger than memory, pass an Arrow Dataset, Scanner, ‘dplyr’ query, RecordBatchReader, or compatible stream to the same high-level writer:

dataset <- arrow::open_dataset("local-parquet-directory")

written <- database$write_table(
  table = "Events",
  data = dataset,
  mapping = "EventsParquet"
)

Here $write_table() calls fabric_kql_write_table(), as in the data-frame example above.

The source is processed in batches instead of first being collected into an R data frame. A supplied RecordBatchReader is single-use.

Export a large KQL result to OneLake

$query() (fabric_kql_query()) is the right interface when the result belongs in R. When the result is too large for the client-result channel or should remain in Fabric, $export() (fabric_kql_export()) runs Kusto’s service-side export and writes the first result set directly to storage:

lakehouse <- fabric_lakehouses("Telemetry workspace")[[1]]

exported <- database$export(
  query = "Events | where observed_at > ago(7d)",
  destination = lakehouse,
  path = "Files/exports/events-weekly",
  format = "parquet",
  name_prefix = "events",
  compression_type = "snappy"
)

exported$state
exported$records
exported$artifacts

The signed-in identity needs write access to the destination. If an export fails, treat any files already written as incomplete and inspect the returned operation before starting a replacement export.