使用 Joda 解析时间戳中的 Druid 错误中的 Parquet 数据摄取

Question

上下文：

我能够从 druid overlord 向 EMR 提交 MapReduce 作业。我的数据源是 Parquet 格式的 S3。时间戳字段值的格式为“2017-09-01 21:14:11:552 IST”。

解析时间戳时出错

问题堆栈跟踪是：

2018-01-18T19:31:52,509 INFO [task-runner-0-priority-0] org.apache.hadoop.mapreduce.Job - Task Id : attempt_1516108443547_0022_m_000068_0, Status : FAILED
Error: io.druid.java.util.common.RE: Failure on row[{"t": "2017-09-01 21:14:11:552 IST"}]
    at io.druid.indexer.HadoopDruidIndexerMapper.map(HadoopDruidIndexerMapper.java:91)
    at io.druid.indexer.DetermineHashedPartitionsJob$DetermineCardinalityMapper.run(DetermineHashedPartitionsJob.java:288)
    ..

Caused by: java.lang.IllegalArgumentException: Invalid format: "2017-09-01 21:14:11:552 IST" is malformed at "IST"
    at org.joda.time.format.DateTimeFormatter.parseDateTime(DateTimeFormatter.java:945)
    at io.druid.java.util.common.parsers.TimestampParser.lambda$createTimestampParser$4(TimestampParser.java:93)
    at io.druid.java.util.common.parsers.TimestampParser.lambda$createObjectTimestampParser$8(TimestampParser.java:129)
    . .

我使用了不同的格式集，可以解析但无法在 joda lib 中获取格式。但是，时间戳格式在 java.text.SimpleDateFormat 中是可读的，请参见以下代码：

用于解析日期的示例 Java 程序

String text = "2017-09-01 21:14:11:552 IST";
SimpleDateFormat sdf =  new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS zzz");
TimeZone gmt = TimeZone.getTimeZone("GMT");
sdf.setTimeZone(gmt);
sdf.setLenient(false);

try {
    Date date = sdf.parse(text);
    System.out.println(date);
    System.out.println(sdf.format(date));
} catch (Exception e) {
    e.printStackTrace();
}

输出

Fri Sep 01 21:14:11 IST 2017
2017-09-01 21:14:11:552 IST

环境：

Druid version: 0.11
EMR version : emr-5.11.0
Hadoop version: Amazon 2.7.3

德鲁伊输入 json

{
  "type": "index_hadoop",
  "spec": {
    "ioConfig": {
      "type": "hadoop",
      "inputSpec": {
        "type": "static",
        "inputFormat": "io.druid.data.input.parquet.DruidParquetInputFormat",
        "paths": "s3://s3_path"
      }
    },
    "dataSchema": {
      "dataSource": "parquet_test1",
      "granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "queryGranularity": "ALL",
        "intervals": ["2017-08-01T00:00:00:000Z/2017-08-02T00:00:00:000Z"]
      },
      "parser": {
        "type": "parquet",
        "parseSpec": {
          "format": "timeAndDims",
          "timestampSpec": {
            "column": "t",
            "format": "yyyy-MM-dd HH:mm:ss:SSS zzz"            
          },
          "dimensionsSpec": {
            "dimensions": [
              "dim1","dim2","dim3"
            ],
            "dimensionExclusions": [],
            "spatialDimensions": []
          }
        }
      },
      "metricsSpec": [{
        "type": "count",
        "name": "count"
      },{
          "type" : "count",
          "name" : "pid",
          "fieldName" : "pid"
        }]
    },
    "tuningConfig": {
      "type": "hadoop",
      "partitionsSpec": {
        "targetPartitionSize": 5000000
      },
      "jobProperties" : {
        "mapreduce.job.user.classpath.first": "true",
        "fs.s3.awsAccessKeyId" : "KEYID",
        "fs.s3.awsSecretAccessKey" : "AccessKey",
        "fs.s3.impl" : "org.apache.hadoop.fs.s3native.NativeS3FileSystem",
        "fs.s3n.awsAccessKeyId" : "KEYID",
        "fs.s3n.awsSecretAccessKey" : "AccessKey",
        "fs.s3n.impl" : "org.apache.hadoop.fs.s3native.NativeS3FileSystem",
        "io.compression.codecs" : "org.apache.hadoop.io.compress.GzipCodec,org.apache.hadoop.io.compress.DefaultCodec,org.apache.hadoop.io.compress.BZip2Codec,org.apache.hadoop.io.compress.SnappyCodec"
      },
      "leaveIntermediate": true
    }
  }, "hadoopDependencyCoordinates": ["org.apache.hadoop:hadoop-client:2.7.3", "org.apache.hadoop:hadoop-aws:2.7.3", "com.hadoop.gplcompression:hadoop-lzo:0.4.20"]
}

可能的解决方案

 1. How to parse "2017-09-01 21:14:11:552 IST" in joda format 

 2. Any config to use SimpleDateFormat for parsing date in timestampSpec, as joda library is used default.

Answer 1

您未能解析时区缩写“IST”。这样的缩写通常是矛盾的。

在这种情况下，“IST”可以代表：“Europe/Dublin”（爱尔兰夏令时间）、“Asia/Jerusalem”（以色列标准时间）、“Asia/Kolkata”（印度标准时间）。看你的名字，我强烈认为你想要印度时间。

现在我讨论几种可能的解决方案及其优缺点。时间库可以使用不同的策略来解决区域名称的歧义。它允许用户明确指定他们想要的区域（用户偏好），或者当前/相关区域设置内的地区/国家信息可用于解析。

乔达时间

唯一！解决方案通过以下代码实现：

String s = "2017-09-01 21:14:11:552 IST";

Map<String, DateTimeZone> preferredJodaZones =
    Collections.singletonMap("IST", DateTimeZone.forID("Asia/Kolkata"));
DateTimeUtils.setDefaultTimeZoneNames(preferredJodaZones); // attention: static (global)
org.joda.time.format.DateTimeFormatter formatter =
    DateTimeFormat.forPattern("yyyy-MM-dd HH:mm:ss:SSS zzz");
DateTime joda = formatter.parseDateTime(s);
System.out.println(joda);
// 2017-09-01T21:14:11.552+05:30

虽然这种基于显式用户偏好的方法可能会满足您的要求，因为您不需要更改依赖项和首选库，但我认为这种方法不是很好，原因有两个：

• 它使用静态方法来设置用户偏好（在多线程环境中可能容易受到攻击）。
• 需要明确知道必须解析哪些区域缩写。

我建议在程序初始化期间只设置一次用户首选项。然后你就可以和 Joda 一起工作了。

---

旧的SimpleDateFormat-class

是的，这适用于你，但不适用于我，因为我机器上的语言环境不是印度。我得到了以色列的时间戳/时间（与印度相差 3.5 小时）。我们看到这个旧类在后台使用关联语言环境的区域信息来解决名称歧义，而不是显式设置的 tz-offset GMT（通过sdf.setTimeZone(gmt);）。

System.out.println(sdf.format(date)); // 2017-09-01 22:14:11:552 IDT

所以请非常小心你的代码在哪里运行。

---

java.time（Java-8 或更高版本）

DateTimeFormatter threeten =
    DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss:SSS zzz", new Locale("en", "IN"));
ZonedDateTime jdt = ZonedDateTime.parse(s, threeten);
System.out.println(jdt);
// 2017-09-01T21:14:11.552+03:00[Asia/Jerusalem] 
// (on my machine! - might work on your machine but is unreliable)

这个实验表明，很遗憾没有使用用于解决 tz-ambiguity 的语言环境信息。但是可以通过基于构建器的方法指定用户偏好：

Set<ZoneId> preferredZones = Collections.singleton(ZoneId.of("Asia/Kolkata"));
DateTimeFormatter threeten2 =
    new DateTimeFormatterBuilder()
    .appendPattern("yyyy-MM-dd HH:mm:ss:SSS ")
    .appendZoneText(TextStyle.SHORT, preferredZones)
    .toFormatter();
ZonedDateTime jdt2 = ZonedDateTime.parse(s, threeten2);
System.out.println(jdt2);
// 2017-09-01T21:14:11.552+05:30[Asia/Kolkata]

在这里，用户偏好可以作为解析器的本地参数给出，并且不会受到任何多线程问题的影响（比 Joda 更好）。

---

Time4J (my lib)

它可以使用类似于 Java-8 的构建器方法来设置用户首选项（此处未显示），也可以在构造格式化程序时部署非固定偏移参数或使用语言环境信息参数（最大灵活性）。

ChronoFormatter<Moment> time4j =
    ChronoFormatter.ofMomentPattern(
        "yyyy-MM-dd HH:mm:ss:SSS zzz",
        PatternType.CLDR,
        new Locale("en", "IN"), // // uses India for resolving tz-ambiguity
        ZonalOffset.UTC 
        // using ASIA.KOLKATA would have higher ranking than locale information
    );
ZonalDateTime zdt = ZonalDateTime.parse(s, time4j); 
// convertible to java.time.ZonedDateTime (zdt.toTemporalAccessor())
System.out.println(zdt);
// 2017-09-01T21:14:11,552+05:30[Asia/Kolkata]