“

系列：S3 数据基座 · 第 23 篇 | 难度：高级 | 阅读时间：20 分钟

数据导出：多格式批量与流式输出

Tags: #DataExport #BatchExport #StreamExport #CSV #Parquet #智策平台

#TL;DR

coomia-dip 平台的数据导出系统支持将 Ontology 数据导出为 CSV、Parquet、Excel、JSON Lines 等多种格式，支持批量导出和流式导出两种模式。本文完整解析数据导出的架构设计，包括导出任务管理、OQL 查询到导出流水线的转换、多格式编码器实现、大数据量分片导出、流式导出（通过 Arrow Flight）、导出安全控制（数据脱敏、权限检查）和导出性能优化。

#1. 导出系统架构

#1.1 整体架构

Code

数据导出系统架构：

┌──────────────────────────────────────────────┐
│              Export Request                    │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐   │
│  │ OQL Query│  │ Entity   │  │ Dashboard│   │
│  │ Export   │  │ Type     │  │ Export   │   │
│  └──────────┘  └──────────┘  └──────────┘   │
└──────────────────────┬───────────────────────┘
                       │
                       ▼
┌──────────────────────────────────────────────┐
│            Export Pipeline                    │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │ Query    │→│ Transform│→│ Encode   │  │
│  │ Execute  │  │ & Mask   │  │ & Write  │  │
│  └──────────┘  └──────────┘  └──────────┘  │
└──────────────────────┬───────────────────────┘
                       │
                       ▼
┌──────────────────────────────────────────────┐
│            Output Targets                     │
│  ┌────┐ ┌────────┐ ┌─────┐ ┌─────┐ ┌────┐ │
│  │CSV │ │Parquet │ │Excel│ │JSONL│ │S3  │ │
│  └────┘ └────────┘ └─────┘ └─────┘ └────┘ │
└──────────────────────────────────────────────┘

#2. 导出 API

#2.1 批量导出

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",
            created_by=request.user,
        )

        # 异步执行
        await self._queue.enqueue(job)
        return job

    async def execute_export(self, job: ExportJob):
        job.status = "RUNNING"

        # 执行 OQL 查询
        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
        job.file_size = await self._storage.get_size(output_path)

# 使用示例
job = await export_service.create_export(ExportRequest(
    query="FETCH Device WHERE region = 'East' ORDER BY name",
    format="csv",
    options=ExportOptions(
        delimiter=",",
        header=True,
        encoding="utf-8-sig",  # Excel 兼容的 UTF-8 BOM
    ),
))

#2.2 流式导出

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. 多格式编码器

#3.1 编码器接口

Python

class ExportEncoder(ABC):
    @abstractmethod
    def encode(self, table: pa.Table) -> bytes: ...

    @abstractmethod
    def encode_batch(self, batch: pa.RecordBatch) -> bytes: ...

    @abstractmethod
    def content_type(self) -> str: ...

    @abstractmethod
    def file_extension(self) -> str: ...


class CSVEncoder(ExportEncoder):
    def __init__(self, options: CSVOptions = None):
        self._options = options or CSVOptions()

    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 content_type(self) -> str:
        return "text/csv"

    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 content_type(self) -> str:
        return "application/octet-stream"

    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 content_type(self) -> str:
        return "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"

    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 content_type(self) -> str:
        return "application/jsonlines"

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

#4. 大数据量分片导出

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),
            'chunk_size': chunk_size,
            'files': output_files,
            'format': encoder.file_extension(),
        }
        manifest_path = f"{base_path}/_manifest.json"
        await self._storage.write(
            manifest_path, json.dumps(manifest).encode()
        )

        return output_files

#5. 数据安全

#5.1 数据脱敏

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,
        'bank_card': lambda v: '****' + v[-4:] if v else v,
    }

    async def apply(self, data: pa.Table, user: str) -> pa.Table:
        user_permissions = await self._auth.get_data_permissions(user)

        for col_name in data.column_names:
            sensitivity = self._get_sensitivity(col_name)
            if sensitivity and not user_permissions.can_see_raw(col_name):
                mask_fn = self.MASKING_RULES.get(sensitivity)
                if mask_fn:
                    data = self._apply_mask_to_column(data, col_name, mask_fn)

        return data

#6. 性能优化

Code

导出性能基准：

┌─────────────────┬──────────┬──────────┬──────────┐
│ 格式 × 数据量     │ 100K 行  │ 1M 行    │ 10M 行   │
├─────────────────┼──────────┼──────────┼──────────┤
│ 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       │
└─────────────────┴──────────┴──────────┴──────────┘

优化策略：
  1. > 100K 行：使用流式导出避免内存溢出
  2. > 1M 行：使用 Parquet 格式（压缩率高、写入快）
  3. > 10M 行：使用 Arrow Flight 直连（最快）
  4. Excel 限制：最大 1M 行（Excel 格式限制）

#7. 测试策略

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)
        # phone 列应被脱敏
        assert '****' in content.decode()

    async def test_chunked_export(self):
        job = await export_service.create_export(ExportRequest(
            query="FETCH Sensor",  # 500K rows
            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

多格式支持覆盖不同使用场景：CSV 用于 Excel 打开和简单集成，Parquet 用于大数据交换，Excel 用于业务人员，JSONL 用于 API 集成。
流式导出解决大数据量内存问题：通过流式编码避免将全部数据加载到内存，支持 10M+ 行的导出。
数据脱敏是导出安全的核心：根据用户权限自动对敏感字段（手机号、身份证等）进行脱敏处理。
分片导出和清单文件支持并行下载：大数据集分片为多个文件，通过清单文件组织，支持客户端并行下载。
Arrow Flight 是高性能场景的首选：相比 CSV/Parquet 文件导出，Arrow Flight 直连传输速度提升 5-10 倍。

#Next Article

下一篇 S3-24《Flight SQL：高性能数据传输协议》 将深入 Arrow Flight SQL 协议的实现细节。

Tags: #DataExport #BatchExport #StreamExport #CSV #Parquet #Excel #JSONL #DataMasking #ArrowFlight #智策平台 #coomia-dip #数据基座