|
|
@@ -13,6 +13,8 @@ import org.apache.parquet.hadoop.ParquetReader;
|
|
|
import org.apache.parquet.hadoop.ParquetWriter;
|
|
|
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
|
|
|
import org.apache.parquet.hadoop.util.HadoopInputFile;
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
|
|
|
import java.io.File;
|
|
|
import java.io.IOException;
|
|
|
@@ -20,63 +22,122 @@ import java.nio.file.Files;
|
|
|
import java.util.HashMap;
|
|
|
import java.util.List;
|
|
|
import java.util.Map;
|
|
|
+import java.util.Objects;
|
|
|
|
|
|
-
|
|
|
+/**
|
|
|
+ * Parquet文件操作工具类
|
|
|
+ * 支持将JSON数据写入Parquet文件、读取Parquet文件最后一行数据
|
|
|
+ * 适配无 ParquetWriter.Mode 的低版本Parquet
|
|
|
+ *
|
|
|
+ * @author 开发者
|
|
|
+ * @date 2025
|
|
|
+ */
|
|
|
public class ParquetUtils {
|
|
|
|
|
|
+ // 日志组件（替换System.out，便于线上排查问题）
|
|
|
+ private static final Logger LOGGER = LoggerFactory.getLogger(ParquetUtils.class);
|
|
|
+
|
|
|
+ // 全局Jackson对象映射器（可复用，提升性能）
|
|
|
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
|
|
|
+
|
|
|
+ // Hadoop配置单例（避免重复创建）
|
|
|
+ private static final Configuration HADOOP_CONF = initHadoopConf();
|
|
|
+
|
|
|
/**
|
|
|
- * 将映射后的JSON数据保存为Parquet文件
|
|
|
+ * 初始化Hadoop配置（适配本地文件系统）
|
|
|
+ * @return 配置好的Hadoop Configuration
|
|
|
*/
|
|
|
- public static void saveAsParquet(
|
|
|
- List<JSONObject> mappedData,
|
|
|
- List<String> standardKeys,
|
|
|
- String outputPath) throws IOException {
|
|
|
+ private static Configuration initHadoopConf() {
|
|
|
+ Configuration conf = new Configuration();
|
|
|
+ conf.set("fs.defaultFS", "file:///");
|
|
|
+ conf.setBoolean("fs.write.checksum", false); // 禁用crc校验文件
|
|
|
+ conf.setInt("io.file.buffer.size", 4096); // 优化文件读写缓冲区
|
|
|
+ return conf;
|
|
|
+ }
|
|
|
|
|
|
- if (mappedData == null || mappedData.isEmpty()) {
|
|
|
- System.out.println("警告: 没有数据需要保存");
|
|
|
+ /**
|
|
|
+ * 将映射后的JSON数据保存为Parquet文件
|
|
|
+ *
|
|
|
+ * @param mappedData 待写入的JSON数据列表
|
|
|
+ * @param standardKeys 标准字段名列表（用于构建Schema）
|
|
|
+ * @param outputPath Parquet文件输出路径
|
|
|
+ * @throws IOException 文件写入失败时抛出异常
|
|
|
+ */
|
|
|
+ public static void saveAsParquet(List<JSONObject> mappedData, List<String> standardKeys, String outputPath) throws IOException {
|
|
|
+ // 参数校验
|
|
|
+ if (Objects.isNull(mappedData) || mappedData.isEmpty()) {
|
|
|
+ LOGGER.warn("待写入的JSON数据为空，无需保存Parquet文件");
|
|
|
return;
|
|
|
}
|
|
|
+ if (Objects.isNull(standardKeys) || standardKeys.isEmpty()) {
|
|
|
+ throw new IllegalArgumentException("标准字段名列表不能为空");
|
|
|
+ }
|
|
|
+ if (Objects.isNull(outputPath) || outputPath.trim().isEmpty()) {
|
|
|
+ throw new IllegalArgumentException("Parquet输出路径不能为空");
|
|
|
+ }
|
|
|
|
|
|
- // 如果文件存在则删除
|
|
|
- File file = new File(outputPath);
|
|
|
- Files.deleteIfExists(file.toPath());
|
|
|
+ // 清理已存在的文件（低版本Parquet无自动覆盖，手动删除）
|
|
|
+ File outputFile = new File(outputPath);
|
|
|
+ if (outputFile.exists()) {
|
|
|
+ Files.deleteIfExists(outputFile.toPath());
|
|
|
+ LOGGER.info("已删除已存在的Parquet文件：{}", outputPath);
|
|
|
+ }
|
|
|
|
|
|
// 创建Avro Schema
|
|
|
Schema schema = createAvroSchema(standardKeys);
|
|
|
|
|
|
- // 创建Parquet写入器
|
|
|
- try (ParquetWriter<GenericRecord> writer = AvroParquetWriter.<GenericRecord>builder(new org.apache.hadoop.fs.Path(outputPath))
|
|
|
+ // 写入Parquet文件（移除Mode，适配低版本）
|
|
|
+ try (ParquetWriter<GenericRecord> writer = AvroParquetWriter.<GenericRecord>builder(new Path(outputPath))
|
|
|
.withSchema(schema)
|
|
|
.withCompressionCodec(CompressionCodecName.SNAPPY)
|
|
|
- .build()) {
|
|
|
+ .withConf(HADOOP_CONF)
|
|
|
+ .build()) { // 移除 withWriteMode，低版本无需指定
|
|
|
|
|
|
- // 将JSON数据转换为Avro记录并写入
|
|
|
+ // 批量写入数据
|
|
|
+ int writeCount = 0;
|
|
|
for (JSONObject jsonObj : mappedData) {
|
|
|
+ if (Objects.isNull(jsonObj)) {
|
|
|
+ LOGGER.warn("跳过空的JSON对象，当前索引：{}", writeCount);
|
|
|
+ continue;
|
|
|
+ }
|
|
|
GenericRecord record = convertToAvroRecord(jsonObj, schema, standardKeys);
|
|
|
writer.write(record);
|
|
|
+ writeCount++;
|
|
|
}
|
|
|
|
|
|
- System.out.println("成功保存Parquet文件: " + outputPath);
|
|
|
- System.out.println("总记录数: " + mappedData.size());
|
|
|
+ LOGGER.info("成功保存Parquet文件：{}，总写入记录数：{}（跳过空记录数：{}）",
|
|
|
+ outputPath, writeCount, mappedData.size() - writeCount);
|
|
|
+
|
|
|
+ } catch (IOException e) {
|
|
|
+ LOGGER.error("写入Parquet文件失败，路径：{}", outputPath, e);
|
|
|
+ throw e; // 抛出异常让上层处理
|
|
|
}
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * 根据标准key创建Avro Schema
|
|
|
+ * 根据标准字段名创建Avro Schema
|
|
|
+ *
|
|
|
+ * @param standardKeys 标准字段名列表
|
|
|
+ * @return Avro Schema对象
|
|
|
*/
|
|
|
private static Schema createAvroSchema(List<String> standardKeys) {
|
|
|
StringBuilder schemaBuilder = new StringBuilder();
|
|
|
- schemaBuilder.append("{\n");
|
|
|
- schemaBuilder.append(" \"type\": \"record\",\n");
|
|
|
- schemaBuilder.append(" \"name\": \"StandardRecord\",\n");
|
|
|
- schemaBuilder.append(" \"namespace\": \"com.example.parquet\",\n");
|
|
|
- schemaBuilder.append(" \"fields\": [\n");
|
|
|
+ schemaBuilder.append("{\n")
|
|
|
+ .append(" \"type\": \"record\",\n")
|
|
|
+ .append(" \"name\": \"StandardRecord\",\n")
|
|
|
+ .append(" \"namespace\": \"com.example.parquet\",\n")
|
|
|
+ .append(" \"fields\": [\n");
|
|
|
|
|
|
for (int i = 0; i < standardKeys.size(); i++) {
|
|
|
String fieldName = standardKeys.get(i);
|
|
|
- schemaBuilder.append(" {\n");
|
|
|
- schemaBuilder.append(" \"name\": \"").append(fieldName).append("\",\n");
|
|
|
- schemaBuilder.append(" \"type\": [\"string\", \"null\"]\n"); // 所有字段都设为string类型，可为null
|
|
|
+ // 字段名合法性校验
|
|
|
+ if (Objects.isNull(fieldName) || fieldName.trim().isEmpty()) {
|
|
|
+ throw new IllegalArgumentException("字段名不能为空，索引：" + i);
|
|
|
+ }
|
|
|
+ schemaBuilder.append(" {\n")
|
|
|
+ .append(" \"name\": \"").append(fieldName.trim()).append("\",\n")
|
|
|
+ .append(" \"type\": [\"string\", \"null\"]\n"); // 所有字段设为string类型，可为null
|
|
|
+
|
|
|
if (i < standardKeys.size() - 1) {
|
|
|
schemaBuilder.append(" },\n");
|
|
|
} else {
|
|
|
@@ -84,71 +145,94 @@ public class ParquetUtils {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- schemaBuilder.append(" ]\n");
|
|
|
- schemaBuilder.append("}");
|
|
|
+ schemaBuilder.append(" ]\n")
|
|
|
+ .append("}");
|
|
|
|
|
|
- return new Schema.Parser().parse(schemaBuilder.toString());
|
|
|
+ try {
|
|
|
+ return new Schema.Parser().parse(schemaBuilder.toString());
|
|
|
+ } catch (Exception e) {
|
|
|
+ LOGGER.error("创建Avro Schema失败，Schema字符串：{}", schemaBuilder, e);
|
|
|
+ throw new RuntimeException("Avro Schema解析失败", e);
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* 将JSONObject转换为Avro GenericRecord
|
|
|
+ *
|
|
|
+ * @param jsonObj JSON对象
|
|
|
+ * @param schema Avro Schema
|
|
|
+ * @param standardKeys 标准字段名列表
|
|
|
+ * @return Avro GenericRecord对象
|
|
|
*/
|
|
|
private static GenericRecord convertToAvroRecord(JSONObject jsonObj, Schema schema, List<String> standardKeys) {
|
|
|
GenericRecord record = new GenericData.Record(schema);
|
|
|
|
|
|
for (String key : standardKeys) {
|
|
|
- Object value = jsonObj.get(key);
|
|
|
+ String fieldName = key.trim();
|
|
|
+ Object value = jsonObj.get(fieldName);
|
|
|
if (value != null) {
|
|
|
- // 将所有值转换为字符串
|
|
|
- record.put(key, value.toString());
|
|
|
+ // 将所有值转换为字符串（兼容各种数据类型）
|
|
|
+ record.put(fieldName, value.toString());
|
|
|
} else {
|
|
|
- record.put(key, null);
|
|
|
+ record.put(fieldName, null);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
return record;
|
|
|
}
|
|
|
|
|
|
-
|
|
|
- // 全局Jackson对象映射器（可复用，提升性能）
|
|
|
- private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
|
|
|
-
|
|
|
/**
|
|
|
- * 读取Parquet文件最后一行并转换为JSON字符串
|
|
|
+ * 读取Parquet文件最后一行并转换为JSON对象
|
|
|
*
|
|
|
* @param parquetFilePath Parquet文件本地路径（如：D:/data/test.parquet）
|
|
|
- * @return 最后一行数据的JSON字符串
|
|
|
- * @throws IOException 读取文件或转换JSON时的异常
|
|
|
+ * @return 最后一行数据的JSON对象（无数据时返回空JSON）
|
|
|
+ * @throws IOException 读取文件失败时抛出异常
|
|
|
*/
|
|
|
public static JSONObject getParquetLastLineAsJson(String parquetFilePath) throws IOException {
|
|
|
- // 1. 初始化Hadoop配置 - 适配本地文件系统（关键：支持file://协议）
|
|
|
- Configuration conf = new Configuration();
|
|
|
- conf.set("fs.defaultFS", "file:///");
|
|
|
- Path parquetPath = new Path(parquetFilePath);
|
|
|
- HadoopInputFile inputFile = HadoopInputFile.fromPath(parquetPath, conf);
|
|
|
+ // 参数校验
|
|
|
+ if (Objects.isNull(parquetFilePath) || parquetFilePath.trim().isEmpty()) {
|
|
|
+ throw new IllegalArgumentException("Parquet文件路径不能为空");
|
|
|
+ }
|
|
|
+ File parquetFile = new File(parquetFilePath);
|
|
|
+ if (!parquetFile.exists()) {
|
|
|
+ LOGGER.error("Parquet文件不存在，路径：{}", parquetFilePath);
|
|
|
+ return new JSONObject();
|
|
|
+ }
|
|
|
+ if (parquetFile.length() == 0) {
|
|
|
+ LOGGER.warn("Parquet文件为空，路径：{}", parquetFilePath);
|
|
|
+ return new JSONObject();
|
|
|
+ }
|
|
|
|
|
|
- // 2. 构建Avro Parquet读取器（Parquet文件多基于Avro序列化）
|
|
|
+ // 构建Hadoop输入文件
|
|
|
+ HadoopInputFile inputFile = HadoopInputFile.fromPath(new Path(parquetFilePath), HADOOP_CONF);
|
|
|
+
|
|
|
+ // 读取Parquet文件，获取最后一行
|
|
|
try (ParquetReader<GenericRecord> reader = AvroParquetReader.<GenericRecord>builder(inputFile)
|
|
|
- .withConf(conf)
|
|
|
+ .withConf(HADOOP_CONF)
|
|
|
.build()) {
|
|
|
|
|
|
- // 3. 遍历Parquet文件，记录最后一行数据（无需缓存所有行，节省内存）
|
|
|
- GenericRecord currentRecord;
|
|
|
GenericRecord lastRecord = null;
|
|
|
+ GenericRecord currentRecord;
|
|
|
+ long lineCount = 0;
|
|
|
+
|
|
|
+ // 遍历文件（仅保留最后一行，节省内存）
|
|
|
while ((currentRecord = reader.read()) != null) {
|
|
|
- lastRecord = currentRecord; // 不断覆盖，最终保留最后一行
|
|
|
+ lastRecord = currentRecord;
|
|
|
+ lineCount++;
|
|
|
}
|
|
|
|
|
|
- // 4. 校验是否存在有效数据
|
|
|
+ LOGGER.info("读取Parquet文件完成，路径：{}，总行数：{}", parquetFilePath, lineCount);
|
|
|
+
|
|
|
+ // 转换为JSON对象
|
|
|
if (lastRecord == null) {
|
|
|
- return new JSONObject(); // 无数据时返回空JSON对象
|
|
|
+ return new JSONObject();
|
|
|
}
|
|
|
-
|
|
|
- // 5. 将GenericRecord转换为Map（方便Jackson转换为JSON）
|
|
|
Map<String, Object> dataMap = convertGenericRecordToMap(lastRecord);
|
|
|
-
|
|
|
- // 6. Jackson将Map转换为格式化的JSON字符串
|
|
|
return new JSONObject(dataMap);
|
|
|
+
|
|
|
+ } catch (IOException e) {
|
|
|
+ LOGGER.error("读取Parquet文件失败，路径：{}", parquetFilePath, e);
|
|
|
+ throw e;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -160,6 +244,10 @@ public class ParquetUtils {
|
|
|
*/
|
|
|
private static Map<String, Object> convertGenericRecordToMap(GenericRecord genericRecord) {
|
|
|
Map<String, Object> resultMap = new HashMap<>();
|
|
|
+ if (Objects.isNull(genericRecord)) {
|
|
|
+ return resultMap;
|
|
|
+ }
|
|
|
+
|
|
|
genericRecord.getSchema().getFields().forEach(field -> {
|
|
|
String fieldName = field.name();
|
|
|
Object fieldValue = genericRecord.get(fieldName);
|
|
|
@@ -168,19 +256,22 @@ public class ParquetUtils {
|
|
|
return resultMap;
|
|
|
}
|
|
|
|
|
|
- // 测试主方法
|
|
|
+ /**
|
|
|
+ * 测试方法
|
|
|
+ * @param args 命令行参数
|
|
|
+ */
|
|
|
public static void main(String[] args) {
|
|
|
try {
|
|
|
- // 替换为你的Parquet文件本地路径
|
|
|
String parquetPath = "C:\\jupyter-lab\\大唐25年科研\\files\\大唐验证数据\\CCWE1500-82.DF\\内蒙塔林花风电场-20251015.parquet";
|
|
|
long begin = System.currentTimeMillis();
|
|
|
JSONObject lastLineJson = getParquetLastLineAsJson(parquetPath);
|
|
|
- System.out.println(System.currentTimeMillis() - begin);
|
|
|
- System.out.println("Parquet文件最后一行数据（JSON格式）：");
|
|
|
- System.out.println(lastLineJson);
|
|
|
+ long costTime = System.currentTimeMillis() - begin;
|
|
|
+
|
|
|
+ LOGGER.info("读取Parquet最后一行耗时：{}ms", costTime);
|
|
|
+ LOGGER.info("Parquet文件最后一行数据（JSON格式）：\n{}", lastLineJson.toJSONString());
|
|
|
+
|
|
|
} catch (Exception e) {
|
|
|
- e.printStackTrace();
|
|
|
+ LOGGER.error("测试读取Parquet文件失败", e);
|
|
|
}
|
|
|
}
|
|
|
-
|
|
|
}
|
wzl