JodaTime在星火groupByKey和countByKey中的应用

Question

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

(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，每个元组都具有相同的date对象。第一个映射生成日期=> 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在date对象上工作：

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

它们是JodaTime的DateTime的实例，它是按预期执行等于。

当我尝试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"]作为值。有些事情正在发生，因为这些值都是列表。

不知何故，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

他说

根据毫秒瞬间和时间顺序，将此对象与指定的对象进行比较，以确保相等。

它们的毫秒都是相等的，它们的时间顺序似乎是：

=> (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没有提供它自己的相等方法并从对象继承它，后者只使用引用相等。

我的理论是，这些日期都是用各自的、不同的、构造的年表对象来反序列化的，但是JodaTime有可能处理这个问题的它自己的序列化器。也许定制的克里奥序列化程序在这方面会有所帮助。

现在，我在火花中使用JodaTime的解决方案是通过调用toInstant或java.util.Date而不是org.joda.time.DateTime来使用org.joda.time .Instant。

两者都涉及丢弃时区信息，这是不理想的，所以如果任何人有更多的信息，这将是非常欢迎！