我们可以阅读Hazelcast Jet中的镶木quet文件吗？

Question

我试图通过Hazelcast读取Parquet文件，因为我在下面写的代码写了，但是Hazelcast提供了任何读取parquet文件的内构建源？

BatchSource<Object> csvData = SourceBuilder
        .batch("parquet-source", x -> {                               
            try {
    
                ParquetReader<GenericData.Record> reader = AvroParquetReader.<GenericData.Record>builder(new Path("D:/test/1651070287920.parquet")).build();
                return reader;
                
            } catch (Exception e) {
                
                return null;
            }
        })
        .<Object>fillBufferFn((reader, buf) -> {                          
            try {
                GenericRecord record;
            if ((record = reader.read()) != null) {
            
                Map<String, String> map = new HashMap<>();
                for (int i = 0; i < headers[0].length; i++) {
                        String value = record.get(i) == null ? "" : record.get(i).toString();
                        map.put(headers[0][i], value);
                }
            
                
                    if (map != null) {
                        rowcount = rowcount + 1; 
                        buf.add(map);
                    }
            } else {
                buf.close();
                return;
            }
            } catch (Exception e) {
                buf.close();
                return;
            }

        })
        .build();

请让我知道Hazelcast Jet中是否已经有任何消息来源。

Answer 1

可以使用 unified File File Connector 。另请参见代码样本。

Answer 2

统一文件连接器

BatchSource<SpecificUser> source = FileSources.files("/data")
  .glob("users.parquet")
  .format(FileFormat.<SpecificUser>parquet())
  .useHadoopForLocalFiles(true)
  .build();

使用 t有一个与您的架构相对应的类，或者您的模式更具动态性，您想返回org.apache.generic.genericrecord从源代码>到map＆lt; string，String＆gt;您可以使用以下内容：

BatchSource<GenericRecord> source = FileSources.files(currentDir + "/target/parquet")
  .glob("file.parquet")
  .option("avro.serialization.data.model", GenericData.class.getName())                                                    
  .useHadoopForLocalFiles(true)
  .format(FileFormat.parquet())
  .build();