[Data] Export dataset operator output schema to event logger by coqian · Pull Request #60086 · ray-project/ray

coqian · 2026-01-13T03:08:57Z

Description

In this PR, we export the output schema of dataset operators so that we can check the output field names and their data types for better observability.

If DataContext.enforce_schemas is set to False, the schema will only be export once for each operator; and if it is set to True, the schema will be exported whenever the fields get updated.

Example export event:

{
  "event_id": "83b3A80eAa283CFFBf",
  "timestamp": 1769111404,
  "source_type": "EXPORT_DATASET_OPERATOR_SCHEMA",
  "event_data": {
    "operator_uuid": "a76411d1-5d28-4027-ad19-56cdcb073410",
    "schema_fields": {
      "int_field": "int64",
      "bool_field": "bool",
      "bytes_field": "binary",
      "string_field": "string",
      "date_field": "date32[day]",
      "datetime_field": "timestamp[us]",
      "numpy_int_field": "int32",
      "numpy_float_field": "double",
      "numpy_array_field": "ArrowTensorType(shape=(3,), dtype=int64)",
      "list_float_field": "list<item: double>",
      "list_list_field": "list<item: list<item: double>>",
      "nested_dict_field": "struct<a: struct<b: string>>",
      "none_field": "null",
    }
  }
}

gemini-code-assist · 2026-01-13T03:09:01Z

Warning

You have reached your daily quota limit. Please wait up to 24 hours and I will start processing your requests again!

python/ray/data/_internal/operator_schema_exporter.py

src/ray/protobuf/export_dataset_operator_schema.proto

python/ray/data/tests/test_operator_schema_export.py

iamjustinhsu

@coqian thanks for doing this!!

Just some questions for my own understanding

Is the event_EXPORT_DATASET_OPERATOR_SCHEMA.log per operator or per dataset?
how do you detect changes in the file?

python/ray/data/_internal/execution/streaming_executor_state.py

iamjustinhsu · 2026-01-27T22:29:31Z

python/ray/data/_internal/operator_schema_exporter.py

@@ -0,0 +1,145 @@
+"""Exporter API for Ray Data operator schema."""


I feel like this file and dataset_state.py can be moved under a new parent directory, thoughts?

oh also, i was thinking -- metadata_exporter.py exports operators too. Can you explain the purposes of this exporter vs. that one?

metadata_exporter is mainly for the dataset and operator metadata and their relationship. After adding operator args and DataContext, the file can be too large to be exported reliably.
The operator_schema_exporter is specifically for the operator schema, as well as potential updates. It can also be large when there are many fields in the schema or it changes many times. And the operator schema is not related with dataset / operator state or other metadata, that's why I create a separate exporter for it.

python/ray/data/_internal/operator_schema_exporter.py

iamjustinhsu · 2026-01-27T22:31:46Z

python/ray/data/_internal/operator_schema_exporter.py

+    return LoggerOperatorSchemaExporter.create_if_enabled()
+
+
+class OperatorSchemaExporter(ABC):


You probably know best, but wondering when we would need this because I can't forsee if we'll need multiple OperatorSchemaExporter

This just follows previous design, such as metadata_exporter.py. Right now we only export to files, but we may have some future export methods, e.g. to a database, web_request, etc. link to previous PR FYI.

python/ray/data/_internal/execution/streaming_executor_state.py

coqian · 2026-01-28T00:19:12Z

@coqian thanks for doing this!!

Just some questions for my own understanding

Is the event_EXPORT_DATASET_OPERATOR_SCHEMA.log per operator or per dataset?

how do you detect changes in the file?

For dataset and operator in the same session, their schema will be written into the same file
We can leverage some other tools to detect file changes, such as vector

src/ray/protobuf/export_dataset_operator_schema.proto

iamjustinhsu

lgtm!

python/ray/data/_internal/operator_schema_exporter.py

iamjustinhsu · 2026-01-29T23:37:47Z

python/ray/data/tests/test_operator_schema_export.py

+    return [json.loads(line) for line in data]
+
+
+def test_export_operator_schema():


up to u, but do u think it would be good to parameterize based on schema types? Here is an example of the ones we use for sanitization. Just wanted to cover all the cases, but no need to do that if u feel it's redundant

This can introduce dependencies on how ToString() of different types are implemented in PyArrow. If it's updated in the future, it may break our test cases here. And different version of them can have different results: an example is the update of datetime64[s] dtype in pandas 2.0, it has different behaviors when we print it before and after 2.0, datetime64[ns] vs datetime64[s]. Our test cases mainly focus on if the schema can be exported to the log file instead.

python/ray/data/_internal/operator_schema_exporter.py

cursor

Cursor Bugbot has reviewed your changes and found 1 potential issue.

src/ray/protobuf/export_dataset_operator_schema.proto

Signed-off-by: cong.qian <[email protected]>

MengjinYan · 2026-01-30T01:31:37Z

@edoakes for review & merge the proto as it seems that core team owns the proto code.

Looks like the PR updates the old export event schema for data events. I think until core support the feature for library events, we will need to keep the export event schema.

…ject#60086) ## Description In this PR, we export the output schema of dataset operators so that we can check the output field names and their data types for better observability. If `DataContext.enforce_schemas` is set to False, the schema will only be export once for each operator; and if it is set to True, the schema will be exported whenever the fields get updated. Example export event: ``` { "event_id": "83b3A80eAa283CFFBf", "timestamp": 1769111404, "source_type": "EXPORT_DATASET_OPERATOR_SCHEMA", "event_data": { "operator_uuid": "a76411d1-5d28-4027-ad19-56cdcb073410", "schema_fields": { "int_field": "int64", "bool_field": "bool", "bytes_field": "binary", "string_field": "string", "date_field": "date32[day]", "datetime_field": "timestamp[us]", "numpy_int_field": "int32", "numpy_float_field": "double", "numpy_array_field": "ArrowTensorType(shape=(3,), dtype=int64)", "list_float_field": "list<item: double>", "list_list_field": "list<item: list<item: double>>", "nested_dict_field": "struct<a: struct<b: string>>", "none_field": "null", } } } ``` Signed-off-by: cong.qian <[email protected]>

…ject#60086) ## Description In this PR, we export the output schema of dataset operators so that we can check the output field names and their data types for better observability. If `DataContext.enforce_schemas` is set to False, the schema will only be export once for each operator; and if it is set to True, the schema will be exported whenever the fields get updated. Example export event: ``` { "event_id": "83b3A80eAa283CFFBf", "timestamp": 1769111404, "source_type": "EXPORT_DATASET_OPERATOR_SCHEMA", "event_data": { "operator_uuid": "a76411d1-5d28-4027-ad19-56cdcb073410", "schema_fields": { "int_field": "int64", "bool_field": "bool", "bytes_field": "binary", "string_field": "string", "date_field": "date32[day]", "datetime_field": "timestamp[us]", "numpy_int_field": "int32", "numpy_float_field": "double", "numpy_array_field": "ArrowTensorType(shape=(3,), dtype=int64)", "list_float_field": "list<item: double>", "list_list_field": "list<item: list<item: double>>", "nested_dict_field": "struct<a: struct<b: string>>", "none_field": "null", } } } ``` Signed-off-by: cong.qian <[email protected]> Signed-off-by: 400Ping <[email protected]>

…ject#60086) ## Description In this PR, we export the output schema of dataset operators so that we can check the output field names and their data types for better observability. If `DataContext.enforce_schemas` is set to False, the schema will only be export once for each operator; and if it is set to True, the schema will be exported whenever the fields get updated. Example export event: ``` { "event_id": "83b3A80eAa283CFFBf", "timestamp": 1769111404, "source_type": "EXPORT_DATASET_OPERATOR_SCHEMA", "event_data": { "operator_uuid": "a76411d1-5d28-4027-ad19-56cdcb073410", "schema_fields": { "int_field": "int64", "bool_field": "bool", "bytes_field": "binary", "string_field": "string", "date_field": "date32[day]", "datetime_field": "timestamp[us]", "numpy_int_field": "int32", "numpy_float_field": "double", "numpy_array_field": "ArrowTensorType(shape=(3,), dtype=int64)", "list_float_field": "list<item: double>", "list_list_field": "list<item: list<item: double>>", "nested_dict_field": "struct<a: struct<b: string>>", "none_field": "null", } } } ``` Signed-off-by: cong.qian <[email protected]> Signed-off-by: Sirui Huang <[email protected]>

## Description In this PR, we export the output schema of dataset operators so that we can check the output field names and their data types for better observability. If `DataContext.enforce_schemas` is set to False, the schema will only be export once for each operator; and if it is set to True, the schema will be exported whenever the fields get updated. Example export event: ``` { "event_id": "83b3A80eAa283CFFBf", "timestamp": 1769111404, "source_type": "EXPORT_DATASET_OPERATOR_SCHEMA", "event_data": { "operator_uuid": "a76411d1-5d28-4027-ad19-56cdcb073410", "schema_fields": { "int_field": "int64", "bool_field": "bool", "bytes_field": "binary", "string_field": "string", "date_field": "date32[day]", "datetime_field": "timestamp[us]", "numpy_int_field": "int32", "numpy_float_field": "double", "numpy_array_field": "ArrowTensorType(shape=(3,), dtype=int64)", "list_float_field": "list<item: double>", "list_list_field": "list<item: list<item: double>>", "nested_dict_field": "struct<a: struct<b: string>>", "none_field": "null", } } } ``` Signed-off-by: cong.qian <[email protected]> Signed-off-by: elliot-barn <[email protected]>

## Description In this PR, we export the output schema of dataset operators so that we can check the output field names and their data types for better observability. If `DataContext.enforce_schemas` is set to False, the schema will only be export once for each operator; and if it is set to True, the schema will be exported whenever the fields get updated. Example export event: ``` { "event_id": "83b3A80eAa283CFFBf", "timestamp": 1769111404, "source_type": "EXPORT_DATASET_OPERATOR_SCHEMA", "event_data": { "operator_uuid": "a76411d1-5d28-4027-ad19-56cdcb073410", "schema_fields": { "int_field": "int64", "bool_field": "bool", "bytes_field": "binary", "string_field": "string", "date_field": "date32[day]", "datetime_field": "timestamp[us]", "numpy_int_field": "int32", "numpy_float_field": "double", "numpy_array_field": "ArrowTensorType(shape=(3,), dtype=int64)", "list_float_field": "list<item: double>", "list_list_field": "list<item: list<item: double>>", "nested_dict_field": "struct<a: struct<b: string>>", "none_field": "null", } } } ``` Signed-off-by: cong.qian <[email protected]>

…ject#60086) ## Description In this PR, we export the output schema of dataset operators so that we can check the output field names and their data types for better observability. If `DataContext.enforce_schemas` is set to False, the schema will only be export once for each operator; and if it is set to True, the schema will be exported whenever the fields get updated. Example export event: ``` { "event_id": "83b3A80eAa283CFFBf", "timestamp": 1769111404, "source_type": "EXPORT_DATASET_OPERATOR_SCHEMA", "event_data": { "operator_uuid": "a76411d1-5d28-4027-ad19-56cdcb073410", "schema_fields": { "int_field": "int64", "bool_field": "bool", "bytes_field": "binary", "string_field": "string", "date_field": "date32[day]", "datetime_field": "timestamp[us]", "numpy_int_field": "int32", "numpy_float_field": "double", "numpy_array_field": "ArrowTensorType(shape=(3,), dtype=int64)", "list_float_field": "list<item: double>", "list_list_field": "list<item: list<item: double>>", "nested_dict_field": "struct<a: struct<b: string>>", "none_field": "null", } } } ``` Signed-off-by: cong.qian <[email protected]> Signed-off-by: Adel Nour <[email protected]>

…ject#60086) ## Description In this PR, we export the output schema of dataset operators so that we can check the output field names and their data types for better observability. If `DataContext.enforce_schemas` is set to False, the schema will only be export once for each operator; and if it is set to True, the schema will be exported whenever the fields get updated. Example export event: ``` { "event_id": "83b3A80eAa283CFFBf", "timestamp": 1769111404, "source_type": "EXPORT_DATASET_OPERATOR_SCHEMA", "event_data": { "operator_uuid": "a76411d1-5d28-4027-ad19-56cdcb073410", "schema_fields": { "int_field": "int64", "bool_field": "bool", "bytes_field": "binary", "string_field": "string", "date_field": "date32[day]", "datetime_field": "timestamp[us]", "numpy_int_field": "int32", "numpy_float_field": "double", "numpy_array_field": "ArrowTensorType(shape=(3,), dtype=int64)", "list_float_field": "list<item: double>", "list_list_field": "list<item: list<item: double>>", "nested_dict_field": "struct<a: struct<b: string>>", "none_field": "null", } } } ``` Signed-off-by: cong.qian <[email protected]> Signed-off-by: peterxcli <[email protected]>

coqian force-pushed the coqian/schema branch 3 times, most recently from 2fc2b9a to fa41ed2 Compare January 15, 2026 08:02

coqian force-pushed the coqian/schema branch from fa41ed2 to e4890dc Compare January 27, 2026 08:20

coqian marked this pull request as ready for review January 27, 2026 08:20

coqian requested review from a team as code owners January 27, 2026 08:20

coqian changed the title ~~[WIP][Data] Export dataset operator output schema to event logger~~ [Data] Export dataset operator output schema to event logger Jan 27, 2026

coqian force-pushed the coqian/schema branch from e4890dc to f4f2702 Compare January 27, 2026 08:23

cursor bot reviewed Jan 27, 2026

View reviewed changes

python/ray/data/_internal/operator_schema_exporter.py Show resolved Hide resolved

src/ray/protobuf/export_dataset_operator_schema.proto Show resolved Hide resolved

coqian force-pushed the coqian/schema branch from f4f2702 to e354fb7 Compare January 27, 2026 10:21

ray-gardener bot added data Ray Data-related issues observability Issues related to the Ray Dashboard, Logging, Metrics, Tracing, and/or Profiling labels Jan 27, 2026

coqian force-pushed the coqian/schema branch from e354fb7 to 9bc0fe2 Compare January 27, 2026 16:09

cursor bot reviewed Jan 27, 2026

View reviewed changes

src/ray/protobuf/export_dataset_operator_schema.proto Show resolved Hide resolved

coqian force-pushed the coqian/schema branch 2 times, most recently from b819b1a to 2452647 Compare January 27, 2026 16:50

cursor bot reviewed Jan 27, 2026

View reviewed changes

python/ray/data/tests/test_operator_schema_export.py Outdated Show resolved Hide resolved

iamjustinhsu reviewed Jan 27, 2026

View reviewed changes

coqian force-pushed the coqian/schema branch from 2452647 to afd947b Compare January 28, 2026 01:26

cursor bot reviewed Jan 28, 2026

View reviewed changes

src/ray/protobuf/export_dataset_operator_schema.proto Show resolved Hide resolved

coqian force-pushed the coqian/schema branch 3 times, most recently from 0b5296e to d89646b Compare January 29, 2026 22:51

cursor bot reviewed Jan 29, 2026

View reviewed changes

src/ray/protobuf/export_dataset_operator_schema.proto Show resolved Hide resolved

iamjustinhsu approved these changes Jan 29, 2026

View reviewed changes

coqian force-pushed the coqian/schema branch from d89646b to 5b23786 Compare January 29, 2026 23:44

cursor bot reviewed Jan 29, 2026

View reviewed changes

python/ray/data/_internal/operator_schema_exporter.py Show resolved Hide resolved

coqian force-pushed the coqian/schema branch from 5b23786 to f8c2d23 Compare January 30, 2026 00:51

cursor bot reviewed Jan 30, 2026

View reviewed changes

src/ray/protobuf/export_dataset_operator_schema.proto Show resolved Hide resolved

[Data] Export dataset operator output schema to event logger

e812bab

Signed-off-by: cong.qian <[email protected]>

coqian force-pushed the coqian/schema branch from f8c2d23 to e812bab Compare January 30, 2026 01:23

edoakes approved these changes Jan 30, 2026

View reviewed changes

edoakes added the go add ONLY when ready to merge, run all tests label Jan 30, 2026

edoakes enabled auto-merge (squash) January 30, 2026 20:21

edoakes merged commit 216182b into ray-project:master Jan 30, 2026
8 checks passed

		@@ -0,0 +1,145 @@
		"""Exporter API for Ray Data operator schema."""

		return LoggerOperatorSchemaExporter.create_if_enabled()


		class OperatorSchemaExporter(ABC):

		return [json.loads(line) for line in data]


		def test_export_operator_schema():

Conversation

coqian commented Jan 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Uh oh!

gemini-code-assist bot commented Jan 13, 2026

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

iamjustinhsu left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

iamjustinhsu Jan 27, 2026

Choose a reason for hiding this comment

Uh oh!

iamjustinhsu Jan 27, 2026

Choose a reason for hiding this comment

Uh oh!

coqian Jan 28, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

iamjustinhsu Jan 27, 2026

Choose a reason for hiding this comment

Uh oh!

coqian Jan 28, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

coqian commented Jan 28, 2026

Uh oh!

Uh oh!

Uh oh!

iamjustinhsu left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

iamjustinhsu Jan 29, 2026

Choose a reason for hiding this comment

Uh oh!

coqian Jan 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

cursor bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

MengjinYan commented Jan 30, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

coqian commented Jan 13, 2026 •

edited

Loading

coqian Jan 30, 2026 •

edited

Loading