发现 flatMap 编译错误：需要 TraversableOnce[String]：TraversableOnce[String]

Question

EDIT#2：这可能与内存有关。日志显示堆外。

是的，肯定与内存有关。 基本上 docker logs 报告所有 来自java的堆外溢出，但jupyter网络笔记本不会将其传递给用户。相反，用户会遇到内核故障和偶尔出现的奇怪行为，例如代码编译不正确。

---

Spark 1.6，尤其是docker run -d .... jupyter/all-spark-notebook

想要统计大约 100 万笔交易的文件中的帐户。

这很简单，它可以在没有火花的情况下完成，但我在尝试使用 spark scala 时遇到了一个奇怪的错误。

输入数据是 RDD[etherTrans] 类型，其中 etherTrans 是包含单个交易的自定义类型：时间戳、来自和至帐户以及以以太币交易的价值。

类 etherTrans(ts_in:Long, afrom_in:String, ato_in:String, ether_in: Float) 扩展可序列化{ var ts: Long = ts_in var afrom: 字符串 = afrom_in var ato: 字符串 = ato_in var ether: 浮点数 = ether_in 覆盖 def toString():String = ts.toString+","+afrom+","+ato+","+ether.toString }

data:RDD[etherTrans] 看起来不错：

data.take(10).foreach(println) etherTrans(1438918233,0xa1e4380a3b1f749673e270229993ee55f35663b4,0x5df9b87991262f6ba471f09758cde1c0fc1de734,3.1337E-14) etherTrans(1438918613,0xbd08e0cddec097db7901ea819a3d1fd9de8951a2,0x5c12a8e43faf884521c2454f39560e6c265a68c8,19.9) etherTrans(1438918630,0x63ac545c991243fa18aec41d4f6f598e555015dc,0xc93f2250589a6563f5359051c1ea25746549f0d8,599.9895) etherTrans(1438918983,0x037dd056e7fdbd641db5b6bea2a8780a83fae180,0x7e7ec15a5944e978257ddae0008c2f2ece0a6090,100.0) etherTrans(1438919175,0x3f2f381491797cc5c0d48296c14fd0cd00cdfa2d,0x4bd5f0ee173c81d42765154865ee69361b6ad189,803.9895) etherTrans(1438919394,0xa1e4380a3b1f749673e270229993ee55f35663b4,0xc9d4035f4a9226d50f79b73aafb5d874a1b6537e,3.1337E-14) etherTrans(1438919451,0xc8ebccc5f5689fa8659d83713341e5ad19349448,0xc8ebccc5f5689fa8659d83713341e5ad19349448,0.0) etherTrans(1438919461,0xa1e4380a3b1f749673e270229993ee55f35663b4,0x5df9b87991262f6ba471f09758cde1c0fc1de734,3.1337E-14) etherTrans(1438919491,0xf0cf0af5bd7d8a3a1cad12a30b097265d49f255d,0xb608771949021d2f2f1c9c5afb980ad8bcda3985,100.0) etherTrans(1438919571,0x1c68a66138783a63c98cc675a9ec77af4598d35e,0xc8ebccc5f5689fa8659d83713341e5ad19349448,50.0)

下一个函数可以解析并以这种方式编写，因为之前的尝试抱怨Array[String] 或List[String] 和TraversableOnce[?] 之间的类型不匹配：

def arrow(e:etherTrans):TraversableOnce[String] = Array(e.afrom,e.ato)

但随后将此函数与 flatMap 一起使用以获取所有帐户的 RDD[String] 失败。

val accts:RDD[String] = data.flatMap(arrow) 名称：编译错误 消息：：38：错误：类型不匹配； 找到：etherTrans（在 $iwC 类中）（在 $iwC 类中）（在 $iwC 类中）（在 $iwC 类中）=> TraversableOnce [String] 必需：etherTrans（在 $iwC 类中）（在 $iwC 类中）（在 $iwC 类中）（在 $iwC 类中）=> TraversableOnce[String] val accts:RDD[String] = data.flatMap(arrow) ^ 堆栈跟踪：

确保您向右滚动以查看它抱怨TraversableOnce[String] 不匹配TraversableOnce[String]

这一定是一个相当普遍的问题，因为在Generate List of Pairs 中出现了更明显的类型不匹配，虽然没有足够的上下文，但在I have a Scala List, how can I get a TraversableOnce? 中提出了建议。

这里发生了什么？

---

编辑：上面报告的问题没有出现，并且代码在较旧的 spark-shell 中运行良好，Spark 1.3.1 在 docker 容器中独立运行。在使用 jupyter/all-spark-notebook docker 容器的 spark 1.6 scala jupyter 环境中运行时会产生错误。

@zero323 还说这个玩具示例：

val rdd = sc.parallelize(Seq((1L, "foo", "bar", 1))).map{ case (ts, fr, to, et) => new etherTrans(ts, fr, to, et) } rdd.flatMap(箭头).collect

在终端 spark-shell 1.6.0/spark 2.10.5 和 Scala 2.11.7 和 Spark 1.5.2 中为他工作。

Answer 1

我认为您应该切换到用例类，它应该可以正常工作。使用“常规”类，在序列化它们时可能会出现奇怪的问题，而且看起来您只需要值对象，因此案例类看起来更适合您的用例。

一个例子：

case class EtherTrans(ts: Long, afrom: String, ato: String, ether: Float)

val source = sc.parallelize(Array(
    (1L, "from1", "to1", 1.234F),
    (2L, "from2", "to2", 3.456F)
))

val data = source.as[EtherTrans]

val data = source.map { l => EtherTrans(l._1, l._2, l._3, l._4) }

def arrow(e: EtherTrans) = Array(e.afrom, e.ato)

data.map(arrow).take(5)
/*
res3: Array[Array[String]] = Array(Array(from1, to1), Array(from2, to2))
*/

data.map(arrow).take(5)
// res3: Array[Array[String]] = Array(Array(from1, to1), Array(from2, to2))

如果需要，您可以创建一些方法/对象来生成案例类。 如果您的逻辑并不真正需要“toString”方法，而只是用于“演示”，请将其保留在案例类之外：您始终可以在存储 if 或显示之前使用 map 操作添加它。

另外，如果您使用的是 Spark 1.6.0 或更高版本，您可以尝试改用 DataSet API，它或多或少看起来像这样：

val data = sqlContext.read.text("your_file").as[EtherTrans]

https://databricks.com/blog/2016/01/04/introducing-spark-datasets.html