Skip to content

Commit 6180bd3

Browse files
maziyarpanahimaziyarpanahi
authored
feat: integrate v2.2 data-processing adapters (#2688)
* Add a Pandas-on-Spark accessor for distributed de-identification (#2156) * feat: add pandas-on-Spark deidentification accessor * fix: finish pandas-on-Spark accessor --------- Co-authored-by: Maziyar Panahi <5762953+maziyarpanahi@users.noreply.github.com> * Add a stream-processor map function and sink for record-level de-identification (#2157) Co-authored-by: Maziyar Panahi <5762953+maziyarpanahi@users.noreply.github.com> * Add an Arrow Flight de-identification service over record batches (#2158) * feat: add Arrow Flight de-identification service * fix: register Arrow Flight guide for publication --------- Co-authored-by: Maziyar Panahi <5762953+maziyarpanahi@users.noreply.github.com> * feat: add Dataflow bundle processor (#2159) * feat: add Dataflow bundle processor * fix: align Dataflow dependency policy * fix: isolate Dataflow runtime dependency --------- Co-authored-by: Maziyar Panahi <5762953+maziyarpanahi@users.noreply.github.com> * Add a Ray Data map-batches de-identification stage with shared model actors (#2160) * feat: add Ray map-batches de-identification * fix: harden Ray batch de-identification --------- Co-authored-by: Maziyar Panahi <5762953+maziyarpanahi@users.noreply.github.com> * Add a distributed SQL engine UDF plugin for in-query de-identification (#2166) * feat: add distributed SQL de-identification UDF * fix: register distributed SQL guide for publication --------- Co-authored-by: Maziyar Panahi <5762953+maziyarpanahi@users.noreply.github.com> * Add PostgreSQL PL/Python de-identification functions (#2167) * feat: add PostgreSQL PL/Python de-identification * fix: harden PostgreSQL de-identification results --------- Co-authored-by: Maziyar Panahi <5762953+maziyarpanahi@users.noreply.github.com> * Add executable UDF column redaction adapter (#2168) * feat: add executable UDF column redaction * fix: harden executable UDF stream results --------- Co-authored-by: Maziyar Panahi <5762953+maziyarpanahi@users.noreply.github.com> * Add a search-engine ingest processor for inline document redaction (#2174) Co-authored-by: Maziyar Panahi <5762953+maziyarpanahi@users.noreply.github.com> --------- Co-authored-by: Maziyar Panahi <5762953+maziyarpanahi@users.noreply.github.com>
1 parent bdbe1c0 commit 6180bd3

32 files changed

Lines changed: 5448 additions & 0 deletions

‎docs/brand/system/publication.yml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -181,13 +181,17 @@ classification:
181181
- security/no-raw-phi-logging.md
182182
- security/tamper-evident-audit-log.md
183183
- integrations-langchain.md
184+
- integrations/arrow-flight.md
184185
- integrations-haystack.md
185186
- integrations-llamaindex.md
186187
- integrations/columnar-redactor.md
187188
- integrations/lakehouse-redaction.md
188189
- integrations/dask.md
190+
- integrations/pandas-on-spark.md
191+
- integrations/ray-map-batches.md
189192
- integrations/sqlalchemy.md
190193
- duckdb-deidentification.md
194+
- integrations/distributed-sql-udf.md
191195
- spark-deidentification.md
192196
- prefect-integration.md
193197
- spacy-component.md

‎docs/integrations/arrow-flight.md‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
# Arrow Flight De-identification
2+
3+
OpenMed's Arrow Flight integration accepts a stream of Arrow record batches,
4+
de-identifies one configured string column, and returns each redacted batch as
5+
soon as it is processed. The service preserves the input schema, row count,
6+
nulls, and every non-target column. It never materializes the complete stream.
7+
8+
Install the optional columnar dependency:
9+
10+
```bash
11+
uv pip install -e ".[columnar]"
12+
```
13+
14+
## Start a Server
15+
16+
```python
17+
from openmed.integrations.arrow_flight import (
18+
ArrowFlightDeidentificationServer,
19+
)
20+
21+
server = ArrowFlightDeidentificationServer(
22+
"grpc://127.0.0.1:8815",
23+
batch_size=512,
24+
)
25+
server.serve()
26+
```
27+
28+
`ArrowFlightDeidentificationServer` uses OpenMed's PII model and
29+
`process_batch(operation="deidentify")` by default. You can set server-wide
30+
defaults with `text_column=` and `policy=`, or select them per exchange in the
31+
Flight command descriptor.
32+
33+
## Exchange Record Batches
34+
35+
```python
36+
import pyarrow as pa
37+
import pyarrow.flight as flight
38+
39+
from openmed.integrations.arrow_flight import make_deidentify_descriptor
40+
41+
client = flight.connect("grpc://127.0.0.1:8815")
42+
descriptor = make_deidentify_descriptor(
43+
"clinical_note",
44+
policy="hipaa_safe_harbor",
45+
)
46+
writer, reader = client.do_exchange(descriptor)
47+
48+
schema = pa.schema(
49+
[
50+
("record_id", pa.int64()),
51+
("clinical_note", pa.string()),
52+
("status", pa.string()),
53+
]
54+
)
55+
writer.begin(schema)
56+
57+
for input_batch in source_batches:
58+
writer.write_batch(input_batch)
59+
output_batch = reader.read_chunk().data
60+
consume(output_batch)
61+
62+
writer.done_writing()
63+
```
64+
65+
The helper creates a versioned JSON `FlightDescriptor` command. For
66+
`DoExchange`, the descriptor is the request-metadata channel; its policy takes
67+
precedence over a server default. Keeping configuration in the descriptor lets
68+
one server apply different OpenMed policy profiles without mixing raw clinical
69+
cells into RPC metadata.
70+
71+
## Privacy and Streaming Contract
72+
73+
- Each incoming `RecordBatch` is passed to `process_batch` and returned before
74+
the server reads the entire stream.
75+
- Only the target string column is converted to Python values for redaction.
76+
All other Arrow arrays are passed through unchanged.
77+
- Response batches retain the input schema and number of rows, including null
78+
placement.
79+
- The service does not log raw or redacted cell values. Errors identify only
80+
the column and row position.
81+
- The command descriptor must not contain patient data. It is only for the
82+
text-column name, policy name, and descriptor version.
83+
84+
## Authentication and TLS Hooks
85+
86+
The server constructor forwards Arrow Flight's `auth_handler`,
87+
`tls_certificates`, `verify_client`, `root_certificates`, and `middleware`
88+
options. These are deployment hooks, not a complete security policy. Production
89+
operators remain responsible for certificate management, authentication design,
90+
network isolation, and authorization.

‎docs/integrations/distributed-sql-udf.md‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
# Distributed SQL De-identification UDF
2+
3+
OpenMed provides a Python entrypoint for distributed SQL UDF bridges that
4+
execute code inside worker processes. The SQL function is logically scalar:
5+
6+
```sql
7+
openmed_deidentify(text VARCHAR, profile VARCHAR) -> VARCHAR
8+
```
9+
10+
The Python bridge should invoke that function in vectorized mode. Each worker
11+
keeps one lazy OpenMed model loader and sends row windows through
12+
`process_batch`, avoiding one model initialization and one inference call per
13+
row. OpenMed does not ship a compiled engine plugin JAR; provisioning and
14+
securing the engine-specific Python bridge remains an operator responsibility.
15+
16+
## Registration descriptor
17+
18+
The engine-neutral descriptor is available as
19+
`OPENMED_DEIDENTIFY_DESCRIPTOR`:
20+
21+
```python
22+
from openmed.integrations.distributed_sql_udf import (
23+
OPENMED_DEIDENTIFY_DESCRIPTOR,
24+
)
25+
26+
descriptor = OPENMED_DEIDENTIFY_DESCRIPTOR
27+
```
28+
29+
Its registration fields are:
30+
31+
```yaml
32+
name: openmed_deidentify
33+
language: python
34+
entrypoint: openmed.integrations.distributed_sql_udf:deidentify_batch
35+
arguments:
36+
- name: text
37+
sql_type: VARCHAR
38+
python_batch: texts
39+
- name: profile
40+
sql_type: VARCHAR
41+
python_batch: profiles
42+
return_type: VARCHAR
43+
vectorized: true
44+
null_handling: called_on_null_input
45+
default_batch_size: 64
46+
```
47+
48+
Map the descriptor to the equivalent fields in the engine's Python UDF
49+
registration system. In particular, configure the bridge to pass arrays of
50+
`text` and `profile` values to `deidentify_batch`; the returned array aligns
51+
one-to-one with the input rows. SQL `NULL` stays `NULL`, and an empty string
52+
stays empty without loading the model.
53+
54+
## Worker lifecycle
55+
56+
For direct integration or an offline registration test, construct the callable
57+
once during worker setup and reuse it for every vector window:
58+
59+
```python
60+
from openmed.integrations.distributed_sql_udf import (
61+
DistributedSQLDeidentifyUDF,
62+
DistributedSQLUDFConfig,
63+
)
64+
65+
openmed_deidentify = DistributedSQLDeidentifyUDF(
66+
config=DistributedSQLUDFConfig(
67+
default_profile="hipaa_safe_harbor",
68+
batch_size=64,
69+
)
70+
)
71+
72+
redacted = openmed_deidentify(
73+
[
74+
"Patient Jane Roe has hypertension.",
75+
None,
76+
"",
77+
],
78+
["hipaa_safe_harbor", None, "hipaa_safe_harbor"],
79+
)
80+
```
81+
82+
The module-level `deidentify_batch` entrypoint uses the same process-local
83+
worker pattern automatically. `deidentify(text, profile)` is also available
84+
for scalar-only bridges, but a bridge that calls Python once per row cannot
85+
benefit from vector batching.
86+
87+
Install the OpenMed model artifact on every worker before accepting queries if
88+
the deployment must remain fully offline. Raw input text is not logged by this
89+
adapter; engine query logs, failure capture, and spill configuration should be
90+
reviewed separately so they do not retain PHI.

‎docs/integrations/pandas-on-spark.md‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,57 @@
1+
# Pandas-on-Spark De-identification
2+
3+
OpenMed registers pandas-on-Spark DataFrame and Series accessors so existing
4+
Pandas-style workflows can redact free-text columns without switching to a
5+
row-at-a-time Spark UDF.
6+
7+
Install the optional Spark extra:
8+
9+
```bash
10+
pip install "openmed[spark]"
11+
```
12+
13+
Import the integration once to register the `.deid` accessors:
14+
15+
```python
16+
import pyspark.pandas as ps
17+
import openmed.integrations.pandas_on_spark # registers the accessors
18+
19+
records = ps.DataFrame(
20+
{
21+
"record_id": ["a", "b"],
22+
"clinical_note": [
23+
"Patient Jane Roe called 555-0100.",
24+
"Follow-up contains no identifiers.",
25+
],
26+
}
27+
)
28+
29+
redacted = records.deid.deidentify(
30+
columns="clinical_note",
31+
policy="hipaa_safe_harbor",
32+
)
33+
print(redacted.to_pandas())
34+
```
35+
36+
The DataFrame method has the same public signature as the local Pandas
37+
accessor: pass one column name or a sequence through `columns`, select a
38+
`method`, and optionally set a policy profile. Extra de-identification options,
39+
including `model_name`, `confidence_threshold`, and language settings, are
40+
forwarded to OpenMed's batch processor.
41+
42+
Series use the same distributed path:
43+
44+
```python
45+
notes = records["clinical_note"].deid.deidentify(
46+
policy="strict_no_leak",
47+
use_safety_sweep=True,
48+
)
49+
```
50+
51+
Under the hood, pandas-on-Spark sends Arrow row groups through a pandas UDF.
52+
OpenMed calls `process_batch` once for each row group and reuses a worker-local
53+
model loader, so the model backbone is not loaded once per row. The integration
54+
does not log source text or emit raw PHI in progress metadata.
55+
56+
This accessor is for the pandas API on Spark. Use the Structured Streaming
57+
helpers for streaming sinks, or the Dask accessor for Dask DataFrames.

‎docs/integrations/ray-map-batches.md‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
# Ray Data map-batches de-identification
2+
3+
OpenMed provides a stateful `Dataset.map_batches` stage for distributed,
4+
columnar de-identification. Ray runs the callable class in an actor pool, and
5+
each actor loads one OpenMed model pipeline in its constructor and reuses that
6+
pipeline for every batch it processes.
7+
8+
Install OpenMed's model dependencies and Ray Data:
9+
10+
```bash
11+
pip install "openmed[hf]" "ray[data]"
12+
```
13+
14+
## Apply the stage
15+
16+
Choose the free-text column explicitly. Non-target columns, null values, batch
17+
row counts, and the dataset's total row count pass through unchanged.
18+
19+
```python
20+
import ray
21+
22+
from openmed.integrations.ray_map_batches import map_batches_deidentify
23+
24+
ray.init()
25+
26+
records = ray.data.from_items(
27+
[
28+
{"record_id": "a", "clinical_note": "Patient Jane Roe called."},
29+
{"record_id": "b", "clinical_note": "No identifiers here."},
30+
]
31+
)
32+
33+
redacted = map_batches_deidentify(
34+
records,
35+
column="clinical_note",
36+
policy_profile="hipaa_safe_harbor",
37+
batch_size=256,
38+
batch_format="pyarrow",
39+
concurrency=4,
40+
)
41+
42+
redacted.write_parquet("/secure/output/redacted-notes")
43+
```
44+
45+
The returned dataset is lazy. A terminal operation such as `write_parquet`,
46+
`materialize`, or `take_all` starts execution.
47+
48+
## Actor pool and batch format
49+
50+
`concurrency=4` creates a fixed pool of four model actors. Use `(minimum,
51+
maximum)` for an autoscaling pool, or `(minimum, maximum, initial)` to set its
52+
initial size as well:
53+
54+
```python
55+
redacted = map_batches_deidentify(
56+
records,
57+
column="clinical_note",
58+
policy_profile="strict_no_leak",
59+
batch_size=512,
60+
batch_format="pandas",
61+
concurrency=(1, 8, 2),
62+
num_cpus=2,
63+
)
64+
```
65+
66+
Supported batch formats are `"pyarrow"` and `"pandas"`. The stage copies the
67+
batch before replacing the target column, leaving Ray's zero-copy input buffers
68+
untouched. `num_cpus`, `num_gpus`, and other Ray worker resource arguments are
69+
forwarded to `Dataset.map_batches`.
70+
71+
## Use the callable class directly
72+
73+
For custom Ray Data plans, pass `RayDeidentifyBatch` to `map_batches`. A class
74+
UDF is important here: a function UDF is stateless and would not retain the
75+
loaded model between calls.
76+
77+
```python
78+
from ray.data import ActorPoolStrategy
79+
80+
from openmed.integrations.ray_map_batches import RayDeidentifyBatch
81+
82+
redacted = records.map_batches(
83+
RayDeidentifyBatch,
84+
batch_size=256,
85+
batch_format="pyarrow",
86+
compute=ActorPoolStrategy(size=4),
87+
fn_constructor_kwargs={
88+
"column": "clinical_note",
89+
"policy_profile": "hipaa_safe_harbor",
90+
},
91+
)
92+
```
93+
94+
Ray's object store and worker processes temporarily hold the input batches.
95+
Deploy this stage only on a trusted, access-controlled cluster appropriate for
96+
the sensitivity of the source data, and write results only to approved storage.

0 commit comments

Comments
 (0)