jodatime - 在 Spark 的 groupByKey 和 countByKey 中使用 JodaTime

Question

我有一个非常简单的 Spark 程序(在 Clojure 中使用 Flambo，但应该很容易理解)。这些都是 JVM 上的对象。我正在测试 local实例(尽管我猜测 Spark 仍然会序列化和反序列化)。

(let [dt (t/date-time 2014)
      input (f/parallelize sc [{:the-date dt :x "A"}
                               {:the-date dt :x "B"}
                               {:the-date dt :x "C"}
                               {:the-date dt :x "D"}])
      by-date (f/map input (f/fn [{the-date :the-date x :x}] [the-date x])))

输入是一个由四个元组组成的 RDD，每个元组都具有相同的日期对象。第一个映射生成日期 => x 的键值 RDD。
input的内容正如预期的那样:

=> (f/foreach input prn)
[#<DateTime 2014-01-01T00:00:00.000Z> "A"]
[#<DateTime 2014-01-01T00:00:00.000Z> "B"]
[#<DateTime 2014-01-01T00:00:00.000Z> "C"]
[#<DateTime 2014-01-01T00:00:00.000Z> "D"]

只是要清楚，平等和 .hashCode处理日期对象:

=> (= dt dt)
true
=> (.hashCode dt)
1260848926
=> (.hashCode dt)
1260848926

它们是 JodaTime 的 DateTime 的实例, 其中 implement equals as expected .

当我尝试时 countByKey ，我得到了预期:

=> (f/count-by-key by-date)
{#<DateTime 2014-01-01T00:00:00.000Z> 4}

但是当我 groupByKey ，好像不行。

=> (f/foreach (f/group-by-key by-date) prn)
[#<DateTime 2014-01-01T00:00:00.000Z> ["A"]]
[#<DateTime 2014-01-01T00:00:00.000Z> ["B"]]
[#<DateTime 2014-01-01T00:00:00.000Z> ["C"]]
[#<DateTime 2014-01-01T00:00:00.000Z> ["D"]]

键都是相同的，所以我希望结果是一个以日期为键和 ["A", "B", "C", "D"] 的条目。作为值(value)。发生了一些事情，因为这些值都是列表。

不知何故 groupByKey没有正确地等同于键。但是 countByKey是。两者有什么区别？我怎样才能让他们表现得一样？

有任何想法吗？

Answer 1

我越来越接近答案了。我认为这属于答案部分而不是问题部分。

这按键分组，变成本地收集，提取第一个项目(日期)。

=> (def result-dates (map first (f/collect (f/group-by-key by-date))))
=> result-dates
(#<DateTime 2014-01-01T00:00:00.000Z>
 #<DateTime 2014-01-01T00:00:00.000Z>
 #<DateTime 2014-01-01T00:00:00.000Z>
 #<DateTime 2014-01-01T00:00:00.000Z>)

哈希码都是一样的

=> (map #(.hashCode %) result-dates)
(1260848926
 1260848926
 1260848926 
 1260848926)

毫秒都是一样的:

=> (map #(.getMillis %) result-dates)
(1388534400000
 1388534400000
 1388534400000
 1388534400000)

equals失败，但 isEquals成功

=> (.isEqual (first result-dates) (second result-dates))
true

=> (.equals (first result-dates) (second result-dates))
false

documentation for .equals says :

Compares this object with the specified object for equality based on the millisecond instant and the Chronology

它们的毫秒数都是相等的，它们的年表似乎是:

=> (map #(.getChronology %) result-dates)
(#<ISOChronology ISOChronology[UTC]>
 #<ISOChronology ISOChronology[UTC]>
 #<ISOChronology ISOChronology[UTC]>
 #<ISOChronology ISOChronology[UTC]>)

然而，年表并不等同。

=> (def a (first result-dates))
=> (def b (second result-dates))

=> (= (.getChronology a) (.getChronology b))
false

虽然哈希码确实

=> (= (.hashCode (.getChronology a)) (.hashCode (.getChronology b)))
true

但是 joda.time.Chronology不提供 its own equals method并从 Object 继承它，它只使用引用相等。

我的理论是这些日期都被反序列化为他们自己的、不同的、构造的 Chronology 对象，但 JodaTime 有 its own serializer这可能处理这个。也许是定制 Kryo序列化程序将在这方面有所帮助。

目前， 我在 Spark 中使用 JodaTime 的解决方案 是使用 org.joda.time .Instant调用 toInstant ，或 java.util.Date而不是 org.joda.time.DateTime .

两者都涉及丢弃时区信息，这并不理想，因此如果有人有更多信息，将非常受欢迎!