gregvenech

Personal

  • Bio
  • Experiments

Professional

  • Projects
  • Resume
  • Work
⇧D 
<!DOCTYPE html><html lang="en" data-app="civic-data-platform" data-theme="dark" data-accessibility="AAA" dat<head><meta charset="utf-8" /><meta name="viewport" content="width=device-width, initial-scale=1, viewport-f<meta name="description" content="Accessible civic data visualization for families and communities across ev<link rel="preconnect" href="https://data.civic.gov" /><link rel="stylesheet" href="/assets/tokens.css" medi<body class="civic-data-platform" data-theme="dark" data-density="comfortable" data-motion="reduced" data-fa<div id="app-root" class="application-container" role="main" aria-label="Civic Data Visualization Platform" <header class="site-header" role="banner" data-sticky="true"><a class="skip-link" href="#main-content">Skip <nav class="primary-navigation" aria-label="Primary navigation" data-collapsible="true" data-expanded="false<ul class="nav-list" role="menubar" aria-orientation="horizontal" data-active-index="0" data-overflow="scrol<li class="nav-item" role="none"><a href="/" class="nav-link" role="menuitem" aria-current="page" data-icon=<li class="nav-item" role="none"><a href="/data" class="nav-link" role="menuitem" data-icon="grid" data-badg<li class="nav-item" role="none"><a href="/historical" class="nav-link" role="menuitem" data-icon="clock" da<li class="nav-item" role="none"><a href="/demographics" class="nav-link" role="menuitem" data-icon="users" <li class="nav-item" role="none"><a href="/about" class="nav-link" role="menuitem" data-icon="info" data-ord</ul></nav></header><main class="main-content" id="main-content" role="main" tabindex="-1" data-scroll-regio<section class="hero-section" aria-labelledby="hero-heading" data-variant="split" data-theme-accent="civic-r<h1 id="hero-heading" class="hero-title" data-animate="fade-up" data-weight="bold">Municipal Record Aggregat<p class="hero-description" data-max-width="60ch">Design systems meet civic data visualization, building acc<div class="hero-actions" role="group" aria-label="Primary actions" data-layout="inline" data-gap="md" data-<button type="button" class="btn btn-primary" aria-label="Explore civic data" data-analytics="cta_explore" d<button type="button" class="btn btn-secondary" aria-label="View historical context" data-analytics="cta_his</div></section><section class="data-visualization-section" aria-labelledby="viz-heading" data-columns="4" d<h2 id="viz-heading" class="section-heading" data-numbered="true" data-weight="semibold">Participation Metri<div class="metrics-grid" role="grid" aria-label="Civic participation metrics" data-sortable="true" data-liv<div class="metric-card" role="gridcell" aria-label="Participation rate" data-trend="up" data-delta="3.2" da<div class="metric-label" id="m-participation">Participation Rate</div><div class="metric-value" data-value=<div class="metric-trend" aria-label="Trending upward three point two percent" data-direction="up" data-peri<div class="metric-card" role="gridcell" aria-label="Total participants" data-trend="up" data-delta="12.5" d<div class="metric-label" id="m-total">Total Participants</div><div class="metric-value" data-value="1247893<div class="metric-trend" aria-label="Trending upward twelve point five percent" data-direction="up" data-pe<div class="metric-card" role="gridcell" aria-label="Event type" data-trend="flat" data-delta="0" data-forma<div class="metric-label" id="m-event">Primary Event Type</div><div class="metric-value" data-value="electio<div class="metric-trend" aria-label="No change from the previous period" data-direction="flat" data-period=<div class="metric-card" role="gridcell" aria-label="Region" data-trend="up" data-delta="4.8" data-format="t<div class="metric-label" id="m-region">Primary Region</div><div class="metric-value" data-value="northeast"<div class="metric-trend" aria-label="Trending upward four point eight percent" data-direction="up" data-per</div></section><section class="table-section" aria-labelledby="table-heading" data-paginated="true" data-pa<h2 id="table-heading" class="section-heading" data-weight="semibold">Recent Records by Municipality and Rep<table class="records-table" role="table" aria-describedby="table-heading" data-sticky-header="true" data-ze<thead><tr><th scope="col" aria-sort="descending" data-key="name">Municipality</th><th scope="col" data-key=<tbody><tr data-row="1"><td>Northeast Corridor</td><td>2026-Q2</td><td class="num">412,908</td><td data-stat<tr data-row="2"><td>Midwest Region</td><td>2026-Q2</td><td class="num">287,551</td><td data-status="pending<tr data-row="3"><td>Western District</td><td>2026-Q2</td><td class="num">198,220</td><td data-status="verif</section><footer class="site-footer" role="contentinfo" data-columns="3"><p class="copyright">Public domain<nav class="footer-nav" aria-label="Footer navigation"><a href="/privacy">Privacy Policy</a><a href="/access</main></div></body></html>
","h":"</main></div></body></html>"}]},{"name":"typescript","lines":[{"p":"import { effect, computed, signal, batch, untrack, type Signal, type ReadonlySignal } from \"@reactive/core\";","h":"import { effect, computed, signal, batch, untrack, type Signal, type ReadonlySignal } from \"@reactive/core\";"},{"p":"import { Scheduler, type Disposable, type EffectFn, type StoreOptions } from \"@reactive/runtime\";","h":"import { Scheduler, type Disposable, type EffectFn, type StoreOptions } from \"@reactive/runtime\";"},{"p":"import { shallowEqual, deepFreeze, invariant, nextTick } from \"@reactive/utils\"; // pure helpers, fully tre","h":"import { shallowEqual, deepFreeze, invariant, nextTick } from \"@reactive/utils\"; // pure helpers, fully tre"},{"p":"type Updater = (prev: Readonly) => Partial; // pure transition applied atomically inside a batched","h":"type Updater<T> = (prev: Readonly<T>) => Partial<T>; // pure transition applied atomically inside a batched"},{"p":"const scheduler = new Scheduler({ flush: \"microtask\", coalesce: true, maxBatch: 512 }); // shared reactive ","h":"const scheduler = new Scheduler({ flush: \"microtask\", coalesce: true, maxBatch: 512 }); // shared reactive "},{"p":"let trackingDepth = 0; // guards against re-entrant effects while an update is already flushing through the","h":"let trackingDepth = 0; // guards against re-entrant effects while an update is already flushing through the"},{"p":"export function createStore(initial: T, options: StoreOptions = {}): Store { // sto","h":"export function createStore<T extends object>(initial: T, options: StoreOptions<T> = {}): Store<T> { // sto"},{"p":"const state = signal(deepFreeze(initial)), version = signal(0); // backing atoms: current value + monoto","h":"const state = signal<T>(deepFreeze(initial)), version = signal(0); // backing atoms: current value + monoto"},{"p":"const subscribers = new Set<(snapshot: Readonly) => void>(); // notified after every committed atomic tr","h":"const subscribers = new Set<(snapshot: Readonly<T>) => void>(); // notified after every committed atomic tr"},{"p":"const equals = options.equals ?? shallowEqual; // pluggable equality gate to skip no-op commits and needles","h":"const equals = options.equals ?? shallowEqual; // pluggable equality gate to skip no-op commits and needles"},{"p":"function setState(patch: Partial | Updater): void { // apply a shallow patch or a pure updater functi","h":"function setState(patch: Partial<T> | Updater<T>): void { // apply a shallow patch or a pure updater functi"},{"p":"const prev = state(), next = typeof patch === \"function\" ? { ...prev, ...patch(prev) } : { ...prev, ...patch","h":"const prev = state(), next = typeof patch === \"function\" ? { ...prev, ...patch(prev) } : { ...prev, ...patch"},{"p":"if (equals(prev, next)) return; // bail out early when nothing meaningfully changed to avoid extra downstre","h":"if (equals(prev, next)) return; // bail out early when nothing meaningfully changed to avoid extra downstre"},{"p":"batch(() => { state.set(deepFreeze(next)); version.set(version() + 1); }); // atomic commit inside a single","h":"batch(() => { state.set(deepFreeze(next)); version.set(version() + 1); }); // atomic commit inside a single"},{"p":"subscribers.forEach((notify) => notify(next as Readonly)); // fan the frozen snapshot out to every liste","h":"subscribers.forEach((notify) => notify(next as Readonly<T>)); // fan the frozen snapshot out to every liste"},{"p":"} // setState: apply a shallow patch or a pure updater, committed atomically to the backing store atoms abo","h":"} // setState: apply a shallow patch or a pure updater, committed atomically to the backing store atoms abo"},{"p":"const derived = computed(() => selectVisible(state())); // memoized projection recomputed only when its inp","h":"const derived = computed(() => selectVisible(state())); // memoized projection recomputed only when its inp"},{"p":"effect(() => { if (trackingDepth === 0) scheduler.enqueue(() => derived()); }); // schedule the recompute r","h":"effect(() => { if (trackingDepth === 0) scheduler.enqueue(() => derived()); }); // schedule the recompute r"},{"p":"function subscribe(fn: (s: Readonly) => void): Disposable { // register a listener, returns a disposer h","h":"function subscribe(fn: (s: Readonly<T>) => void): Disposable { // register a listener, returns a disposer h"},{"p":"subscribers.add(fn); fn(state()); // eagerly push the current snapshot so new subscribers start fully in sy","h":"subscribers.add(fn); fn(state()); // eagerly push the current snapshot so new subscribers start fully in sy"},{"p":"return { dispose: () => void subscribers.delete(fn) }; // idempotent teardown removes the listener from the","h":"return { dispose: () => void subscribers.delete(fn) }; // idempotent teardown removes the listener from the"},{"p":"} // subscribe: hands back a disposer; every listener lives in the set and is notified after each atomic co","h":"} // subscribe: hands back a disposer; every listener lives in the set and is notified after each atomic co"},{"p":"return { state, version, setState, derived, subscribe } as Store; // the store's minimal, immutable publ","h":"return { state, version, setState, derived, subscribe } as Store<T>; // the store's minimal, immutable publ"},{"p":"} // createStore returns its public api; every internal atom stays encapsulated inside this factory's closu","h":"} // createStore returns its public api; every internal atom stays encapsulated inside this factory's closu"},{"p":"export const selectVisible = (s: T): T[] => Object.values(s as object).filter(Boolean) as T[]; // strip","h":"export const selectVisible = <T,>(s: T): T[] => Object.values(s as object).filter(Boolean) as T[]; // strip"},{"p":"export function computedMap(source: ReadonlySignal>): ReadonlySignal> { // derive","h":"export function computedMap<K, V>(source: ReadonlySignal<Map<K, V>>): ReadonlySignal<Map<K, V>> { // derive"},{"p":"return computed(() => new Map([...source()].sort(([, a], [, b]) => Number(b) - Number(a)))); // keep the ma","h":"return computed(() => new Map([...source()].sort(([, a], [, b]) => Number(b) - Number(a)))); // keep the ma"},{"p":"} // computedMap: exposes a derived, descending-sorted projection of the source map, recomputed lazily on d","h":"} // computedMap: exposes a derived, descending-sorted projection of the source map, recomputed lazily on d"},{"p":"export function untrackedRead(sig: ReadonlySignal): T { return untrack(() => sig()); } // read withou","h":"export function untrackedRead<T>(sig: ReadonlySignal<T>): T { return untrack(() => sig()); } // read withou"},{"p":"class EffectScope implements Disposable { // collects child effects and disposables for deterministic clean","h":"class EffectScope implements Disposable { // collects child effects and disposables for deterministic clean"},{"p":"private readonly children: Disposable[] = []; // disposed in strict reverse registration order on the teard","h":"private readonly children: Disposable[] = []; // disposed in strict reverse registration order on the teard"},{"p":"add(child: Disposable): void { invariant(child, \"child must be disposable\"); this.children.push(child); } /","h":"add(child: Disposable): void { invariant(child, \"child must be disposable\"); this.children.push(child); } /"},{"p":"run(fn: (scope: this) => R): R { trackingDepth++; try { return fn(this); } finally { trackingDepth--; } }","h":"run<R>(fn: (scope: this) => R): R { trackingDepth++; try { return fn(this); } finally { trackingDepth--; } }"},{"p":"dispose(): void { while (this.children.length) this.children.pop()!.dispose(); } // tear down every registe","h":"dispose(): void { while (this.children.length) this.children.pop()!.dispose(); } // tear down every registe"},{"p":"} // EffectScope: nested reactive effects are disposed together, in strict reverse order, for deterministic","h":"} // EffectScope: nested reactive effects are disposed together, in strict reverse order, for deterministic"},{"p":"export type Store = { state: Signal; version: Signal; setState(p: Partial | Updater): vo","h":"export type Store<T> = { state: Signal<T>; version: Signal<number>; setState(p: Partial<T> | Updater<T>): vo"},{"p":"export const SIGNAL_BRAND: unique symbol = Symbol(\"reactive.signal\"); // nominal brand used to identify sig","h":"export const SIGNAL_BRAND: unique symbol = Symbol(\"reactive.signal\"); // nominal brand used to identify sig"},{"p":"scheduler.on(\"flush\", () => { if (trackingDepth === 0) scheduler.drain(); }); // drain queued effects every","h":"scheduler.on(\"flush\", () => { if (trackingDepth === 0) scheduler.drain(); }); // drain queued effects every"},{"p":"export function withScope(run: (scope: EffectScope) => R): R { // run a unit of work inside a disposable","h":"export function withScope<R>(run: (scope: EffectScope) => R): R { // run a unit of work inside a disposable"},{"p":"const scope = new EffectScope(); // any effects created during the run are collected and then torn down tog","h":"const scope = new EffectScope(); // any effects created during the run are collected and then torn down tog"},{"p":"try { return scope.run(run); } finally { scope.dispose(); } // guarantee cleanup even when the callback thr","h":"try { return scope.run(run); } finally { scope.dispose(); } // guarantee cleanup even when the callback thr"},{"p":"} // withScope: runs a unit of work and always disposes every effect the callback created, throw or return","h":"} // withScope: runs a unit of work and always disposes every effect the callback created, throw or return"},{"p":"const DEFAULT_OPTIONS: StoreOptions = { equals: shallowEqual, scheduler, freeze: true } as const; /","h":"const DEFAULT_OPTIONS: StoreOptions<object> = { equals: shallowEqual, scheduler, freeze: true } as const; /"},{"p":"export default createStore; // the default export is the store factory, the primary entry point for the who","h":"export default createStore; // the default export is the store factory, the primary entry point for the who"},{"p":"export { effect, computed, signal, batch, untrack, EffectScope, computedMap, withScope, selectVisible, nextT","h":"export { effect, computed, signal, batch, untrack, EffectScope, computedMap, withScope, selectVisible, nextT"},{"p":"// build: reactive build --target es2022 --format esm,cjs --treeshake && publish the package to the public n","h":"// build: reactive build --target es2022 --format esm,cjs --treeshake && publish the package to the public n"}]},{"name":"python","lines":[{"p":"from __future__ import annotations # async ETL pipeline: extract from sources, transform in windows, load t","h":"from __future__ import annotations # async ETL pipeline: extract from sources, transform in windows, load t"},{"p":"import asyncio, logging, time # stdlib primitives for concurrency, structured logging, and high-resolution ","h":"import asyncio, logging, time # stdlib primitives for concurrency, structured logging, and high-resolution "},{"p":"from collections import defaultdict, deque, namedtuple # buffering, windowed aggregation, and lightweight r","h":"from collections import defaultdict, deque, namedtuple # buffering, windowed aggregation, and lightweight r"},{"p":"from pipeline.io import AsyncSource, ParquetSink, RetryPolicy, backoff_jitter, checkpoint # pluggable io ad","h":"from pipeline.io import AsyncSource, ParquetSink, RetryPolicy, backoff_jitter, checkpoint # pluggable io ad"},{"p":"log = logging.getLogger(\"pipeline.etl\") # structured logs shipped to the central collector for every pipeli","h":"log = logging.getLogger(\"pipeline.etl\") # structured logs shipped to the central collector for every pipeli"},{"p":"MAX_INFLIGHT, WINDOW_SIZE, RETRY_ATTEMPTS = 64, 10_000, 5 # bound concurrency, window depth, and the retry ","h":"MAX_INFLIGHT, WINDOW_SIZE, RETRY_ATTEMPTS = 64, 10_000, 5 # bound concurrency, window depth, and the retry "},{"p":"async def extract(source: AsyncSource, *, batch_size: int = 500) -> \"asyncio.Queue[list | None]\": # pipelin","h":"async def extract(source: AsyncSource, *, batch_size: int = 500) -> \"asyncio.Queue[list | None]\": # pipelin"},{"p":"queue: asyncio.Queue = asyncio.Queue(maxsize=MAX_INFLIGHT) # a bounded queue applies backpressure to the pr","h":"queue: asyncio.Queue = asyncio.Queue(maxsize=MAX_INFLIGHT) # a bounded queue applies backpressure to the pr"},{"p":"async for page in source.paginate(size=batch_size): # stream the remote pages lazily rather than buffering ","h":"async for page in source.paginate(size=batch_size): # stream the remote pages lazily rather than buffering "},{"p":"rows = [row for row in page if row.get(\"valid\", True)] # a cheap prefilter drops obviously invalid records ","h":"rows = [row for row in page if row.get(\"valid\", True)] # a cheap prefilter drops obviously invalid records "},{"p":"await queue.put(rows) # hand the filtered page to the transform stage, blocking when the queue is already a","h":"await queue.put(rows) # hand the filtered page to the transform stage, blocking when the queue is already a"},{"p":"log.debug(\"extracted page\", extra={\"rows\": len(rows), \"source\": source.name, \"queued\": queue.qsize()}) # tr","h":"log.debug(\"extracted page\", extra={\"rows\": len(rows), \"source\": source.name, \"queued\": queue.qsize()}) # tr"},{"p":"if queue.full(): await asyncio.sleep(backoff_jitter()) # yield cooperatively when the consumer stage falls ","h":"if queue.full(): await asyncio.sleep(backoff_jitter()) # yield cooperatively when the consumer stage falls "},{"p":"await queue.put(None) # a sentinel value signals the transform stage that the source stream has been fully ","h":"await queue.put(None) # a sentinel value signals the transform stage that the source stream has been fully "},{"p":"return queue # the populated queue is handed directly to transform() as the input for the next stage in the","h":"return queue # the populated queue is handed directly to transform() as the input for the next stage in the"},{"p":"async def transform(queue: \"asyncio.Queue\") -> dict[str, list]: # stage two: dedupe, normalize, window reco","h":"async def transform(queue: \"asyncio.Queue\") -> dict[str, list]: # stage two: dedupe, normalize, window reco"},{"p":"buckets: dict[str, deque] = defaultdict(lambda: deque(maxlen=WINDOW_SIZE)) # sliding windows keyed by the r","h":"buckets: dict[str, deque] = defaultdict(lambda: deque(maxlen=WINDOW_SIZE)) # sliding windows keyed by the r"},{"p":"seen: set = set() # tracks composite natural keys already processed so duplicates within a single window ar","h":"seen: set = set() # tracks composite natural keys already processed so duplicates within a single window ar"},{"p":"dropped = 0 # counts records rejected as duplicates, surfaced later as a data-quality metric in the run sum","h":"dropped = 0 # counts records rejected as duplicates, surfaced later as a data-quality metric in the run sum"},{"p":"while (batch := await queue.get()) is not None: # drain the queue until the extract stage finally posts its","h":"while (batch := await queue.get()) is not None: # drain the queue until the extract stage finally posts its"},{"p":"for record in batch: # normalize every record, dedupe it by composite key, and route it into a per-region w","h":"for record in batch: # normalize every record, dedupe it by composite key, and route it into a per-region w"},{"p":"key = (record[\"region\"], record[\"metric\"], record[\"period\"]) # composite natural key used for deduplication","h":"key = (record[\"region\"], record[\"metric\"], record[\"period\"]) # composite natural key used for deduplication"},{"p":"if key in seen: dropped += 1; continue # skip any records we have already ingested during the current windo","h":"if key in seen: dropped += 1; continue # skip any records we have already ingested during the current windo"},{"p":"seen.add(key); buckets[record[\"region\"]].append(normalize(record)) # accept and append the normalized recor","h":"seen.add(key); buckets[record[\"region\"]].append(normalize(record)) # accept and append the normalized recor"},{"p":"log.info(\"transform complete\", extra={\"buckets\": len(buckets), \"dropped\": dropped, \"unique\": len(seen)}) # ","h":"log.info(\"transform complete\", extra={\"buckets\": len(buckets), \"dropped\": dropped, \"unique\": len(seen)}) # "},{"p":"return {region: list(window) for region, window in buckets.items()} # snapshot each sliding window as a pla","h":"return {region: list(window) for region, window in buckets.items()} # snapshot each sliding window as a pla"},{"p":"def normalize(record: dict) -> dict: # coerce raw source records into the warehouse's canonical, typed colu","h":"def normalize(record: dict) -> dict: # coerce raw source records into the warehouse's canonical, typed colu"},{"p":"return {**record, \"value\": round(float(record.get(\"v\", 0.0)), 2), \"ingested_at\": time.time()} # stamp and r","h":"return {**record, \"value\": round(float(record.get(\"v\", 0.0)), 2), \"ingested_at\": time.time()} # stamp and r"},{"p":"async def load(sink: ParquetSink, partitions: dict[str, list]) -> int: # stage three: write the partitions ","h":"async def load(sink: ParquetSink, partitions: dict[str, list]) -> int: # stage three: write the partitions "},{"p":"written = 0 # a running total of the rows flushed to the columnar sink across every regional partition in t","h":"written = 0 # a running total of the rows flushed to the columnar sink across every regional partition in t"},{"p":"for region, rows in sorted(partitions.items(), key=lambda kv: kv[0]): # deterministic, stable partition wri","h":"for region, rows in sorted(partitions.items(), key=lambda kv: kv[0]): # deterministic, stable partition wri"},{"p":"written += await sink.write_partition(region, rows, compression=\"zstd\", checkpoint=checkpoint(region)) # fl","h":"written += await sink.write_partition(region, rows, compression=\"zstd\", checkpoint=checkpoint(region)) # fl"},{"p":"log.debug(\"wrote partition\", extra={\"region\": region, \"rows\": len(rows), \"total\": written}) # per-partition","h":"log.debug(\"wrote partition\", extra={\"region\": region, \"rows\": len(rows), \"total\": written}) # per-partition"},{"p":"return written # the caller emits this count as the run's rows_loaded metric in the final run summary telem","h":"return written # the caller emits this count as the run's rows_loaded metric in the final run summary telem"},{"p":"class PipelineError(RuntimeError): ... # raised when any stage exhausts its retry budget and can no longer ","h":"class PipelineError(RuntimeError): ... # raised when any stage exhausts its retry budget and can no longer "},{"p":"STAGES: tuple[str, ...] = (\"extract\", \"transform\", \"load\", \"verify\", \"publish\") # the ordered pipeline stag","h":"STAGES: tuple[str, ...] = (\"extract\", \"transform\", \"load\", \"verify\", \"publish\") # the ordered pipeline stag"},{"p":"RunSummary = namedtuple(\"RunSummary\", \"rows_in rows_out dropped duration_s\") # immutable record for the run","h":"RunSummary = namedtuple(\"RunSummary\", \"rows_in rows_out dropped duration_s\") # immutable record for the run"},{"p":"policy = RetryPolicy(max_attempts=RETRY_ATTEMPTS, base_delay=0.25, jitter=backoff_jitter) # a shared io ret","h":"policy = RetryPolicy(max_attempts=RETRY_ATTEMPTS, base_delay=0.25, jitter=backoff_jitter) # a shared io ret"},{"p":"async def run(source: AsyncSource, sink: ParquetSink) -> RunSummary: # orchestrate the full extract-transfo","h":"async def run(source: AsyncSource, sink: ParquetSink) -> RunSummary: # orchestrate the full extract-transfo"},{"p":"started = time.perf_counter() # capture a monotonic start time so the total run duration is measured accura","h":"started = time.perf_counter() # capture a monotonic start time so the total run duration is measured accura"},{"p":"queue = await extract(source) # kick off ingestion, populating the bounded queue that the transform stage c","h":"queue = await extract(source) # kick off ingestion, populating the bounded queue that the transform stage c"},{"p":"partitions = await transform(queue) # dedupe and window the streamed records into per-region partition list","h":"partitions = await transform(queue) # dedupe and window the streamed records into per-region partition list"},{"p":"rows = await load(sink, partitions) # flush every partition to the columnar warehouse and count the rows wr","h":"rows = await load(sink, partitions) # flush every partition to the columnar warehouse and count the rows wr"},{"p":"duration = round(time.perf_counter() - started, 3) # total wall-clock seconds elapsed for the entire pipeli","h":"duration = round(time.perf_counter() - started, 3) # total wall-clock seconds elapsed for the entire pipeli"},{"p":"return RunSummary(rows_in=source.count, rows_out=rows, dropped=0, duration_s=duration) # telemetry for the ","h":"return RunSummary(rows_in=source.count, rows_out=rows, dropped=0, duration_s=duration) # telemetry for the "},{"p":"if __name__ == \"__main__\": asyncio.run(run(AsyncSource.env(), ParquetSink.env())) # entry point for the bat","h":"if __name__ == \"__main__\": asyncio.run(run(AsyncSource.env(), ParquetSink.env())) # entry point for the bat"}]}]
Advertisement
Advertisement