“

Series: S3 Data Foundation · Article 23 | Level: Advanced | Reading Time: 20 min

Data Export: Multi-Format Batch and Streaming Output

Tags: #DataExport #BatchExport #StreamExport #CSV #Parquet #coomia-dip

#TL;DR

The coomia-dip data export system supports exporting Ontology data to CSV, Parquet, Excel, JSON Lines, and other formats, with both batch and streaming export modes. This article fully dissects the data export architecture, including export job management, OQL query to export pipeline conversion, multi-format encoder implementation, large-data chunked export, streaming export (via Arrow Flight), export security controls (data masking, permission checks), and export performance optimization.

#1. Export System Architecture

Code

Data Export System Architecture:

+----------------------------------------------+
|              Export Request                    |
|  +----------+  +----------+  +----------+    |
|  | OQL Query|  | Entity   |  | Dashboard|    |
|  | Export   |  | Type     |  | Export   |    |
|  +----------+  +----------+  +----------+    |
+-----------------------+-----------------------+
                        |
                        v
+----------------------------------------------+
|            Export Pipeline                     |
|  +----------+  +----------+  +----------+    |
|  | Query    |->| Transform|->| Encode   |    |
|  | Execute  |  | & Mask   |  | & Write  |    |
|  +----------+  +----------+  +----------+    |
+-----------------------+-----------------------+
                        |
                        v
+----------------------------------------------+
|            Output Targets                     |
|  +----+ +--------+ +-----+ +-----+ +----+   |
|  |CSV | |Parquet | |Excel| |JSONL| |S3  |   |
|  +----+ +--------+ +-----+ +-----+ +----+   |
+----------------------------------------------+

#2. Export API

#2.1 Batch Export

Python

class ExportService:
    async def create_export(self, request: ExportRequest) -> ExportJob:
        await self._auth.check_export_permission(
            request.user, request.entity_type
        )

        job = ExportJob(
            id=uuid4().hex,
            query=request.query,
            format=request.format,
            status="PENDING",
        )
        await self._queue.enqueue(job)
        return job

    async def execute_export(self, job: ExportJob):
        job.status = "RUNNING"
        result = await self._query_engine.execute(job.query)
        masked = await self._masking.apply(result, job.created_by)

        encoder = self._get_encoder(job.format)
        output_path = f"exports/{job.id}/{job.id}.{job.format}"

        if masked.num_rows > self._config.large_export_threshold:
            await self._export_chunked(masked, encoder, output_path)
        else:
            content = encoder.encode(masked)
            await self._storage.write(output_path, content)

        job.status = "COMPLETED"
        job.output_path = output_path
        job.row_count = masked.num_rows

#2.2 Streaming Export

Python

class StreamingExportService:
    async def stream_export(
        self, request: ExportRequest
    ) -> AsyncIterator[bytes]:
        result_stream = self._query_engine.execute_streaming(
            request.query
        )
        encoder = self._get_encoder(request.format)

        if request.format == 'csv':
            yield encoder.encode_header(request.columns)

        async for batch in result_stream:
            masked = await self._masking.apply_batch(
                batch, request.user
            )
            yield encoder.encode_batch(masked)

#3. Multi-Format Encoders

Python

class CSVEncoder(ExportEncoder):
    def encode(self, table: pa.Table) -> bytes:
        sink = io.BytesIO()
        csv.write_csv(table, sink, write_options=csv.WriteOptions(
            delimiter=self._options.delimiter,
            include_header=self._options.header,
        ))
        return sink.getvalue()

    def file_extension(self) -> str:
        return "csv"


class ParquetEncoder(ExportEncoder):
    def encode(self, table: pa.Table) -> bytes:
        sink = io.BytesIO()
        pq.write_table(table, sink, compression='snappy')
        return sink.getvalue()

    def file_extension(self) -> str:
        return "parquet"


class ExcelEncoder(ExportEncoder):
    def encode(self, table: pa.Table) -> bytes:
        df = table.to_pandas()
        sink = io.BytesIO()
        with pd.ExcelWriter(sink, engine='openpyxl') as writer:
            df.to_excel(writer, index=False, sheet_name='Data')
        return sink.getvalue()

    def file_extension(self) -> str:
        return "xlsx"


class JSONLinesEncoder(ExportEncoder):
    def encode(self, table: pa.Table) -> bytes:
        lines = []
        for batch in table.to_batches():
            for row in batch.to_pylist():
                lines.append(json.dumps(row, ensure_ascii=False))
        return '\n'.join(lines).encode('utf-8')

    def file_extension(self) -> str:
        return "jsonl"

#4. Large Data Chunked Export

Python

class ChunkedExporter:
    async def export_chunked(
        self, data: pa.Table, encoder: ExportEncoder,
        base_path: str, chunk_size: int = 100_000
    ) -> list[str]:
        output_files = []
        total_rows = data.num_rows

        for i in range(0, total_rows, chunk_size):
            chunk = data.slice(i, min(chunk_size, total_rows - i))
            chunk_path = (
                f"{base_path}/part-{i:08d}.{encoder.file_extension()}"
            )
            content = encoder.encode(chunk)
            await self._storage.write(chunk_path, content)
            output_files.append(chunk_path)

        manifest = {
            'total_rows': total_rows,
            'chunk_count': len(output_files),
            'files': output_files,
        }
        await self._storage.write(
            f"{base_path}/_manifest.json",
            json.dumps(manifest).encode()
        )
        return output_files

#5. Data Security

Python

class DataMaskingEngine:
    MASKING_RULES = {
        'phone': lambda v: v[:3] + '****' + v[-4:] if v else v,
        'email': lambda v: v[0] + '***@' + v.split('@')[1] if '@' in v else v,
        'id_card': lambda v: v[:6] + '********' + v[-4:] if v else v,
        'name': lambda v: v[0] + '*' * (len(v) - 1) if v else v,
    }

    async def apply(self, data: pa.Table, user: str) -> pa.Table:
        permissions = await self._auth.get_data_permissions(user)
        for col in data.column_names:
            sensitivity = self._get_sensitivity(col)
            if sensitivity and not permissions.can_see_raw(col):
                mask_fn = self.MASKING_RULES.get(sensitivity)
                if mask_fn:
                    data = self._apply_mask(data, col, mask_fn)
        return data

#6. Performance Benchmarks

Code

Export Performance Benchmarks:

+-----------------+----------+----------+----------+
| Format x Size    | 100K rows| 1M rows  | 10M rows |
+-----------------+----------+----------+----------+
| CSV              | 0.5s     | 4s       | 40s      |
| Parquet          | 0.3s     | 2s       | 15s      |
| Excel            | 2s       | 15s      | OOM      |
| JSONL            | 0.8s     | 7s       | 65s      |
| Arrow Flight     | 0.1s     | 0.8s     | 6s       |
+-----------------+----------+----------+----------+

Optimization strategies:
  1. > 100K rows: streaming export to avoid OOM
  2. > 1M rows: Parquet format (high compression, fast write)
  3. > 10M rows: Arrow Flight direct connection (fastest)
  4. Excel limit: max 1M rows (Excel format limitation)

#7. Testing Strategy

Python

class TestDataExport:

    async def test_csv_export_correctness(self):
        job = await export_service.create_export(ExportRequest(
            query="FETCH Person LIMIT 100", format="csv"
        ))
        await wait_for_job(job.id)
        content = await storage.read(job.output_path)
        lines = content.decode().strip().split('\n')
        assert len(lines) == 101  # header + 100 rows

    async def test_parquet_roundtrip(self):
        original = await query("FETCH Device LIMIT 1000")
        job = await export_service.create_export(ExportRequest(
            query="FETCH Device LIMIT 1000", format="parquet"
        ))
        await wait_for_job(job.id)
        exported = pq.read_table(job.output_path)
        assert original.num_rows == exported.num_rows

    async def test_data_masking_applied(self):
        job = await export_service.create_export(ExportRequest(
            query="FETCH Customer SELECT name, phone LIMIT 10",
            format="csv",
        ))
        await wait_for_job(job.id)
        content = await storage.read(job.output_path)
        assert '****' in content.decode()

    async def test_chunked_export(self):
        job = await export_service.create_export(ExportRequest(
            query="FETCH Sensor", format="parquet"
        ))
        await wait_for_job(job.id)
        manifest = json.loads(
            await storage.read(f"{job.output_dir}/_manifest.json")
        )
        assert manifest['chunk_count'] > 1

#Key Takeaways

Multi-format support covers different use cases: CSV for Excel and simple integration, Parquet for big data exchange, Excel for business users, JSONL for API integration.
Streaming export solves large-data memory issues: streaming encoding avoids loading all data into memory, supporting 10M+ row exports.
Data masking is the core of export security: automatically masks sensitive fields (phone numbers, IDs) based on user permissions.
Chunked export with manifest files enables parallel downloads: large datasets split into multiple files, organized by manifest for client-side parallel downloading.
Arrow Flight is the top choice for high-performance scenarios: compared to CSV/Parquet file export, Arrow Flight direct transfer is 5-10x faster.

#Next Article

Next up: S3-24 "Flight SQL: High-Performance Data Transfer Protocol" will dive into the implementation details of the Arrow Flight SQL protocol.

Tags: #DataExport #BatchExport #StreamExport #CSV #Parquet #Excel #JSONL #DataMasking #ArrowFlight #coomia-dip #DataFoundation