使用Dataflow向BQ导入数据时特殊变音字符显示异常问题

问题核心 #

使用Dataflow将API数据导入BigQuery时，DataflowRunner运行模式下数据出现转义字符（如正确内容Zählerablesung显示为转义形式），但DirectRunner本地运行无此问题；已尝试调整UTF-8编码但无效。

代码问题排查 #

你提供的代码中有一处明显无效且会引发异常的逻辑：

output = new String(output.getBytes("UTF-8"));

这行代码在output未初始化时就调用getBytes，会触发空指针异常，且InputStreamReader已指定StandardCharsets.UTF_8，读取的字符串本身已是正确编码，无需额外转码。

根本原因 #

直接将JSON对象转为字符串输出，在Dataflow分布式运行环境（DataflowRunner）中，序列化/传输环节会对字符串中的特殊字符（如Unicode字符）进行二次转义；而DirectRunner本地运行时序列化逻辑更简单，未触发该问题。

修复方案 #

1. 移除无效编码转换代码 #

直接删除output = new String(output.getBytes("UTF-8"));这一行。

2. 输出结构化数据而非JSON字符串 #

不要输出字符串形式的JSON，而是将JSON解析为结构化对象（如Beam的Row或自定义POJO），让Dataflow正确序列化数据，避免BQ写入时的转义：

方案一：使用Beam Row对象（推荐，贴合BQ表结构）

@ProcessElement
public void processElement(OutputReceiver<Row> receiver) throws IOException, ParseException {
    if (this.conn.getResponseCode() == 200) {
        BufferedReader in = new BufferedReader(new InputStreamReader(this.conn.getInputStream(), StandardCharsets.UTF_8));
        StringBuilder response = new StringBuilder();
        String output;
        while ((output = in.readLine()) != null) {
            response.append(output);
        }
        in.close();
        JSONArray jsonArray = new JSONArray(response.toString());
        
        // 提前定义与BQ表匹配的Schema
        Schema yourSchema = Schema.builder()
            .addStringField("target_field") // 对应存储Zählerablesung的字段
            // 添加其他字段
            .build();
        
        for (int i = 0; i < jsonArray.length(); i++) {
            JSONObject jsonObj = jsonArray.getJSONObject(i);
            Row row = Row.withSchema(yourSchema)
                .addValue(jsonObj.getString("your_json_key")) // 替换为实际JSON键名
                // 依次添加其他字段值
                .build();
            receiver.output(row);
        }
    }
}

后续使用BigQueryIO.writeTableRows()将Row写入BQ，确保Schema与BQ表完全匹配。

方案二：使用自定义POJO类

// 定义与JSON结构匹配的POJO
public class ApiData {
    private String targetField; // 对应Zählerablesung的字段
    // 添加其他字段及Getter/Setter
}

@ProcessElement
public void processElement(OutputReceiver<ApiData> receiver) throws IOException, ParseException {
    if (this.conn.getResponseCode() == 200) {
        ObjectMapper mapper = new ObjectMapper();
        BufferedReader in = new BufferedReader(new InputStreamReader(this.conn.getInputStream(), StandardCharsets.UTF_8));
        
        JSONArray jsonArray = mapper.readValue(in, JSONArray.class);
        for (Object obj : jsonArray) {
            ApiData data = mapper.convertValue(obj, ApiData.class);
            receiver.output(data);
        }
        in.close();
    }
}

写入BQ时使用BigQueryIO.write().to("dataset.table").withFormatFunction(...)将POJO转换为BQ可识别的格式。

3. 验证BQ写入配置 #

确保BQ表的字段类型为STRING（对应存储特殊字符的字段），且Dataflow的写入逻辑未开启额外的转义配置。

内容的提问来源于stack exchange，提问作者MI Haque