Method write

dataflow/utils/storage.py:389–432

Write data to Myscale/ClickHouse table. data: pd.DataFrame or List[dict]，每行是data字段内容（dict）。

(self, data: Any)

            raise ValueError(f"Unsupported output type: {output_type}")

    def write(self, data: Any) -> Any:
        """
        Write data to Myscale/ClickHouse table.
        data: pd.DataFrame or List[dict]，每行是data字段内容（dict）。
        """
        if isinstance(data, list):
            df = pd.DataFrame(data)
        elif isinstance(data, pd.DataFrame):
            df = data
        else:
            raise ValueError(f"Unsupported data type: {type(data)}")
        # data字段本身就是每行的内容
        if 'data' not in df.columns:
            # 兼容直接传入dict列表的情况
            df['data'] = df.apply(lambda row: row.to_dict(), axis=1)
        # 统一处理data列
        df['data'] = df['data'].apply(lambda x: x if isinstance(x, dict) else (json.loads(x) if isinstance(x, str) else {}))
        # 自动填充pipeline_id, task_id, raw_data_id, min_hashes
        df['pipeline_id'] = self.pipeline_id
        df['task_id'] = self.output_task_id
        df['raw_data_id'] = df['data'].apply(lambda d: d.get(SYS_FIELD_PREFIX + 'raw_data_id', 0) if isinstance(d, dict) else 0)
        df['min_hashes'] = df['data'].apply(lambda d: _default_min_hashes(d) if isinstance(d, dict) else [0])
        # data字段转为JSON字符串
        df['data'] = df['data'].apply(lambda x: json.dumps(x, ensure_ascii=False) if not isinstance(x, str) else x)
        # 只保留必需字段
        required_cols = ['pipeline_id', 'task_id', 'raw_data_id', 'min_hashes', 'data']
        df = df[required_cols]
        records = df.to_dict(orient="records")
        values = [
            (
                rec['pipeline_id'],
                rec['task_id'],
                int(rec['raw_data_id']),
                rec['min_hashes'],
                rec['data']
            ) for rec in records
        ]
        insert_sql = f"""
        INSERT INTO {self.table} (pipeline_id, task_id, raw_data_id, min_hashes, data)
        VALUES
        """
        self.logger.info(f"Inserting {len(values)} rows into {self.table}")
        self.client.execute(insert_sql, values)
        return f"Inserted {len(values)} rows into {self.table}"