【问题标题】:How to save parquet file in hdfs without spark or framework?如何在没有火花或框架的情况下将镶木地板文件保存在 hdfs 中?
【发布时间】:2023-03-20 11:05:01
【问题描述】:

我想使用 java 将 parquet 文件直接保存到 hdfs。

这是我用来生成 parquet 文件并将它们存储在本地的代码,但现在我想将它们存储在 hdfs 中。

 final String schemaLocation = "/home/javier/FlinkProjects/kafka-flink/src/main/java/com/grallandco/demos/avro.json";
  final Schema avroSchema = new Schema.Parser().parse(new File(schemaLocation));
  final MessageType parquetSchema = new AvroSchemaConverter().convert(avroSchema);
  final WriteSupport writeSupport = new AvroWriteSupport(parquetSchema, avroSchema);
  final String parquetFile = "/home/javier/parquet-files/data" + postfijoFilename + ".parquet";
  final Path path = new Path(parquetFile);
  AvroParquetWriter parquetWriter = new AvroParquetWriter(path,
          avroSchema, CompressionCodecName.SNAPPY, ParquetWriter.DEFAULT_BLOCK_SIZE, ParquetWriter.DEFAULT_PAGE_SIZE);
  final GenericRecord record = new GenericData.Record(avroSchema);
  record.put(Constantes.CAMPO_ID, datos[0]);
  record.put("movie", datos[1]);
  record.put("date", datos[2]);
  record.put("imdb", datos[3]);
  parquetWriter.write(record);

我想替换这个

 final String parquetFile = "/home/javier/parquet-files/data" + postfijoFilename + ".parquet";

有一个 hadoop hdfs 路径,有什么想法吗???

【问题讨论】:

  • 好吧。您需要一个 Hadoop FileSystem 对象。那有几种写文件的方法

标签: java hadoop hdfs parquet


【解决方案1】:

您可以通过以下方式执行此操作(请注意,该位置必须存在并在代码中更改您的 hdfsurl 和用户名。可能需要在 hdfs 中有架构):

final String schemaLocation = "/home/javier/FlinkProjects/kafka-flink/src/main/java/com/grallandco/demos/avro.json";
final Schema avroSchema = new Schema.Parser().parse(new File(schemaLocation));
final MessageType parquetSchema = new AvroSchemaConverter().convert(avroSchema);
final WriteSupport writeSupport = new AvroWriteSupport(parquetSchema,  avroSchema);
final Path path = new Path("/user/hduser/parquet-files/data" + 
       postfijoFilename + ".parquet");
Configuration configuration = new Configuration();
String hdfsUrl = "hdfs://hadoopnamenode:9000/";
String username = "hduser";
FileSystem fs= FileSystem.get(new URI(hdfsUrl), configuration);
UserGroupInformation ugi = 
UserGroupInformation.createRemoteUser(username);
ugi.doAs(new PrivilegedExceptionAction<Void>() {
            public Void run() throws Exception { 
                AvroParquetWriter parquetWriter = new 
                    AvroParquetWriter(path,
                    avroSchema, 
                    CompressionCodecName.SNAPPY,                
                    ParquetWriter.DEFAULT_BLOCK_SIZE, 
                    ParquetWriter.DEFAULT_PAGE_SIZE);
                final GenericRecord record = new 
                    GenericData.Record(avroSchema);
                record.put(Constantes.CAMPO_ID, datos[0]);
                record.put("movie", datos[1]);
                record.put("date", datos[2]);
                record.put("imdb", datos[3]);
                parquetWriter.write(record);
                return null;
                }
    });

【讨论】:

    猜你喜欢
    • 2017-03-17
    • 1970-01-01
    • 1970-01-01
    • 2016-01-18
    • 2019-05-07
    • 1970-01-01
    • 2020-01-10
    • 2019-10-28
    • 2018-10-14
    相关资源
    最近更新 更多