当前位置: 首页 > 面试题库 >

如何使用ParquetWriter将TIMESTAMP逻辑类型(INT96)写入Parquet?

益锦程
2023-03-14
问题内容

我有一个使用org.apache.parquet.hadoop.ParquetWriter的工具将CSV数据文件转换为镶木地板数据文件。

目前,它只能处理int32doublestring

我需要支持实木复合地板timestamp逻辑类型(标注为int96),但由于无法在网上找到精确的规格,因此我不知道该怎么做。

似乎这种时间戳记编码(int96)很少,不能很好地支持。我在网上发现很少的规格详细信息。这个github自述文件指出:

保存为int96的时间戳记由一天(前8个字节)和儒略日(后4个字节)中的纳秒组成。

特别:

  1. 我对MessageType模式中的列使用哪种镶木地板类型?我假设我应该使用原始类型,但是我不确定是否可以指定逻辑类型?PrimitiveTypeName.INT96
  2. 如何写数据?即我要以哪种格式将时间戳记写入小组?对于INT96时间戳,我假设我必须写一些二进制类型?

这是我的代码的简化版本,演示了我正在尝试做的事情。具体来说,请看一下“ TODO”注释,这是代码中与上述问题相关的两点。

List<Type> fields = new ArrayList<>();
fields.add(new PrimitiveType(Type.Repetition.OPTIONAL, PrimitiveTypeName.INT32, "int32_col", null));
fields.add(new PrimitiveType(Type.Repetition.OPTIONAL, PrimitiveTypeName.DOUBLE, "double_col", null));
fields.add(new PrimitiveType(Type.Repetition.OPTIONAL, PrimitiveTypeName.STRING, "string_col", null));

// TODO: 
//   Specify the TIMESTAMP type. 
//   How? INT96 primitive type? Is there a logical timestamp type I can use w/ MessageType schema?
fields.add(new PrimitiveType(Type.Repetition.OPTIONAL, PrimitiveTypeName.INT96, "timestamp_col", null));

MessageType schema = new MessageType("input", fields);

// initialize writer
Configuration configuration = new Configuration();
configuration.setQuietMode(true);
GroupWriteSupport.setSchema(schema, configuration);
ParquetWriter<Group> writer = new ParquetWriter<Group>(
  new Path("output.parquet"),
  new GroupWriteSupport(),
  CompressionCodecName.SNAPPY,
  ParquetWriter.DEFAULT_BLOCK_SIZE,
  ParquetWriter.DEFAULT_PAGE_SIZE,
  1048576,
  true,
  false,
  ParquetProperties.WriterVersion.PARQUET_1_0,
  configuration
);

// write CSV data
CSVParser parser = CSVParser.parse(new File(csv), StandardCharsets.UTF_8, CSVFormat.TDF.withQuote(null));
ArrayList<String> columns = new ArrayList<>(schemaMap.keySet());
int colIndex;
int rowNum = 0;
for (CSVRecord csvRecord : parser) {
  rowNum ++;
  Group group = f.newGroup();
  colIndex = 0;
  for (String record : csvRecord) {
    if (record == null || record.isEmpty() || record.equals( "NULL")) {
      colIndex++;
      continue;
    }


    record = record.trim();
    String type = schemaMap.get(columns.get(colIndex)).get("type").toString();
    MessageTypeConverter.addTypeValueToGroup(type, record, group, colIndex++);

    switch (colIndex) {
      case 0: // int32
        group.add(colIndex, Integer.parseInt(record));
        break;
      case 1: // double
        group.add(colIndex, Double.parseDouble(record));
        break;
      case 2: // string
        group.add(colIndex, record);
        break;
      case 3:
        // TODO: convert CSV string value to TIMESTAMP type (how?)
        throw new NotImplementedException();
    }
  }
  writer.write(group);
}
writer.close();

问题答案:

我通过使用来自spark
sql的这段代码作为参考来弄清楚了。

INT96二进制编码分为两部分:前8个字​​节为自午夜以来的纳秒,最后4个字节为儒略日

String value = "2019-02-13 13:35:05";

final long NANOS_PER_HOUR = TimeUnit.HOURS.toNanos(1);
final long NANOS_PER_MINUTE = TimeUnit.MINUTES.toNanos(1);
final long NANOS_PER_SECOND = TimeUnit.SECONDS.toNanos(1);

// Parse date
SimpleDateFormat parser = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
cal.setTime(parser.parse(value));

// Calculate Julian days and nanoseconds in the day
LocalDate dt = LocalDate.of(cal.get(Calendar.YEAR), cal.get(Calendar.MONTH)+1, cal.get(Calendar.DAY_OF_MONTH));
int julianDays = (int) JulianFields.JULIAN_DAY.getFrom(dt);
long nanos = (cal.get(Calendar.HOUR_OF_DAY) * NANOS_PER_HOUR)
        + (cal.get(Calendar.MINUTE) * NANOS_PER_MINUTE)
        + (cal.get(Calendar.SECOND) * NANOS_PER_SECOND);

// Write INT96 timestamp
byte[] timestampBuffer = new byte[12];
ByteBuffer buf = ByteBuffer.wrap(timestampBuffer);
buf.order(ByteOrder.LITTLE_ENDIAN).putLong(nanos).putInt(julianDays);

// This is the properly encoded INT96 timestamp
Binary tsValue = Binary.fromReusedByteArray(timestampBuffer);


 类似资料:
  • 我已经将Luxon创建的datetime值保存到postgres数据库列中,类型为TIMESTAMP(3)。然后我想使用这个值,把它转换成其他时区,等等。然而,我似乎想不出如何“利用”它。 我使用以下方法创建了该对象: 然后,我将其插入到postgres数据库中,插入时间戳(3)类型的列中。 我使用查询从数据库中提取它。当我记录它时,它说它的值是: 我可以找到很多关于如何将日期插入sql的教程,但

  • 主要内容:Oracle TIMESTAMP数据类型简介,Oracle TIMESTAMP文字,Oracle TIMESTAMP示例,格式化TIMESTAMP值,提取TIMESTAMP组件,默认的TIMESTAMP格式在本教程中将学习Oracle 数据类型以及如何在Oracle数据库中有效处理数据。 Oracle TIMESTAMP数据类型简介 数据类型用于存储日期和时间数据,包括年,月,日,时,分和秒。 另外,它存储小数秒,它不是由DATE数据类型存储的。 要定义列,请使用以下语法: 指定字段小

  • 1个2个 如图1所示,在我的逻辑中,我有两种类型的代理,所以我设置了3个源。然而,当我运行该模型时,它无法通过队列和后续步骤,我认为这是因为代理类型的设置。您知道如何设置可以使3个不同的代理通过此逻辑的代理类型吗?

  • 问题内容: 最新的Avro编译器(1.8.2)使用基于Joda-Time的实现为日期逻辑类型生成Java源。如何配置Avro编译器以生成使用Java 8日期时间API的源? 问题答案: 目前(avro 1.8.2)这是不可能的。它被硬编码以生成Joda日期/时间类。 当前分支已切换到Java 8,并且存在一个未解决的问题(使用Pull Request )来添加生成具有类型的类的功能。 不幸的是,我

  • 若要添加一个新的实体,点击工具栏的 “实体”按钮,并点击画布的任意位置。你可以从浏览器的模型选项卡添加一个现有的实体,简单地从模型选项卡拖放实体到画布。 如果图表符号设置为默认, 图标代表属性为一个主键。而 图标则代表属性为一个索引。 【注意】如果你按住 Control 键并点按属性,你可以选择添加、插入、删除、重命名属性及设置属性为主键。 在画布中实体对象的弹出式菜单选项包括: 选项 描述 设计

  • 问题内容: 我正在尝试使用Python检查回文。我拥有的代码是非常for循环的。 在我看来,当人们从C转到Python时,最大的错误就是尝试使用Python实现C逻辑,这使事情运行缓慢,并且只是没有充分利用该语言。 我在这个网站上看到了。搜索“ C-style for”,即Python没有C-style for循环。可能已经过时,但是我将其解释为意味着Python具有自己的方法。 我尝试环顾四周,