2

我正在尝试使用数据流将一些数据从 pub sub 移动到云存储。我需要为 scio / beam 提供时间戳，以便它可以将数据分组到窗口中。

我有一个简单的案例类来模拟我的事件，它看起来像这样（删除了一些字段）

case class DataEvent(source: String,                      
                     record: AnyRef,
                     timestampUtc: DateTime,
                     publishedUtc: DateTime)

我的管道由此开始。pub sub 中的事件为 json 格式，我使用 json4s 反序列化：

sc
                 .pubsubSubscription[String]("subscription")
                 .map(event => parse(event).camelizeKeys.extract[DataEvent])
                 .timestampBy({ event => event.timestampUtc.toInstant })

在同一范围内，我定义了 Json4s 的隐式格式

 implicit val formats: Formats = new DefaultFormats {
      override def dateFormatter = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")
    } ++ org.json4s.ext.JodaTimeSerializers.all

我正在使用支持 jodatime 的 json4s.ext，请注意案例类中的日期时间是 joda 时间。这个扩展库似乎有一些问题，因为我得到以下异常：

java.lang.IllegalArgumentException: unable to serialize anonymous function map@{PubSubToGcsJob.scala:78}
java.lang.IllegalArgumentException: unable to serialize anonymous function map@{PubSubToGcsJob.scala:78}
    at org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:53)
    at org.apache.beam.sdk.util.SerializableUtils.clone(SerializableUtils.java:90)
    at org.apache.beam.sdk.transforms.ParDo$SingleOutput.<init>(ParDo.java:591)
    at org.apache.beam.sdk.transforms.ParDo.of(ParDo.java:435)
    at com.spotify.scio.values.PCollectionWrapper$class.parDo(PCollectionWrapper.scala:58)
    at com.spotify.scio.values.SCollectionImpl.parDo(SCollection.scala:1181)
    at com.spotify.scio.values.SCollection$class.map(SCollection.scala:359)
    at com.spotify.scio.values.SCollectionImpl.map(SCollection.scala:1181)
    at PubSubToGcsJob$.main(PubSubToGcsJob.scala:78)
    at PubSubToGcsJob.main(PubSubToGcsJob.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
Caused by: java.io.NotSerializableException: org.json4s.ext.IntervalSerializer$$anon$1
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1184)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
    at scala.collection.immutable.List$SerializationProxy.writeObject(List.scala:468)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at java.io.ObjectStreamClass.invokeWriteObject(ObjectStreamClass.java:1028)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1496)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
    at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
    at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
    at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
    at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
    at org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:49)
    at org.apache.beam.sdk.util.SerializableUtils.clone(SerializableUtils.java:90)
    at org.apache.beam.sdk.transforms.ParDo$SingleOutput.<init>(ParDo.java:591)
    at org.apache.beam.sdk.transforms.ParDo.of(ParDo.java:435)
    at com.spotify.scio.values.PCollectionWrapper$class.parDo(PCollectionWrapper.scala:58)
    at com.spotify.scio.values.SCollectionImpl.parDo(SCollection.scala:1181)
    at com.spotify.scio.values.SCollection$class.map(SCollection.scala:359)
    at com.spotify.scio.values.SCollectionImpl.map(SCollection.scala:1181)
    at PubSubToGcsJob$.main(PubSubToGcsJob.scala:78)
    at PubSubToGcsJob.main(PubSubToGcsJob.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)

我尝试了一项工作，我在其中创建了 timestampUtc 和 publishedUtc 字符串的类型，并像这样解析管道内的字符串：

val formatter = DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'")

and in the pipeline

                 .timestampBy({ event => formatter.parseDateTime(event.timestampUtc).toInstant })

但我得到一个类似的例外：

java.lang.IllegalArgumentException: unable to serialize org.apache.beam.sdk.transforms.WithTimestamps$AddTimestampsDoFn@7b7a1a8f
java.lang.IllegalArgumentException: unable to serialize org.apache.beam.sdk.transforms.WithTimestamps$AddTimestampsDoFn@7b7a1a8f
    at org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:53)
    at org.apache.beam.sdk.util.SerializableUtils.clone(SerializableUtils.java:90)
... etc

为什么会发生这种情况，我该如何解决？

谢谢

4

1 回答 1

2

当 lambda 函数从其闭包中拉入不可序列化的内容时，就会发生这种情况。在这种情况下，我怀疑val formatter。

一种解决方法是将 移动val到一个同伴对象，以便它在工作人员上静态初始化，不需要通过 ser/de。例如：

object Util { val formatter = ... }

有关详细信息，请参阅： https ://www.lyh.me/lambda-serialization.html

于 2018-09-21T09:39:23.633 回答