如何在 Apache Spark 中从 Twitter 推文中提取主题标签（或其他数组）

Question

我正在尝试使用 Apache Spark 从 JSON Tweet 对象文件中对 Twitter Tweet 数据进行分析。

这是我使用 Spark 的 jsonFile 方法加载它的方式：

val sqc = new org.apache.spark.sql.SQLContext(sc)

val tweets = sqc.jsonFile("stored_tweets/*.json")

tweets.registerTempTable("tweets")

接下来，我仅对带有以下行的主题标签实体进行抽样：

val hashtags = sqc.sql("SELECT entities.hashtags FROM tweets LIMIT 3")

hashtags.take(1) 

结果是：

res14: 数组[org.apache.spark.sql.Row] = Array([ArrayBuffer([ArrayBuffer(43, 50),在线], [ArrayBuffer(51, 61),营销], [ArrayBuffer(88, 102),growthhacking], [ArrayBuffer(103, 111),入站], [ArrayBuffer(112, 120),启动], [ArrayBuffer(121, 138),内容营销])])

如果你仔细观察，数据就在那里，但它被包裹在 Array([ArrayBuffer(xx,yy), hashtag) 中。

有些人建议使用 .map() 或 .flatMap() 或使用 .getAs() 方法的一些自定义函数，但我不明白这应该如何工作。

有什么想法吗？

5 月 23 日更新：

一直在浏览 Spark 文档。仍然没有进展。 Spark SQL Row 文档 (https://spark.apache.org/docs/1.0.1/api/java/org/apache/spark/sql/api/java/Row.html) 建议使用诸如

之类的代码

import org.apache.spark.sql._
val row = hashtags.take(1)
row(0)

但是，在这种情况下，这会产生

res124: org.apache.spark.sql.Row = [ArrayBuffer([ArrayBuffer(43, 50),online], [ArrayBuffer(51, 61),marketing], [ArrayBuffer(88, 102),growthhacking], [ArrayBuffer(103, 111),inbound], [ArrayBuffer(112, 120),启动], [ArrayBuffer(121, 138),内容营销])]

在这里，这个 StackOverflow 帖子 (org.apache.spark.sql.Row to Int) 建议使用 .get() 系列方法，例如 .getString() 但我的尝试没有t产生很多结果：

row(0).getString(0)

产量：

java.lang.ClassCastException: scala.collection.mutable.ArrayBuffer 无法转换为 java.lang.String

还有，

row(0).getString(1)

产量：

:28: 错误：值 getString 不是 Any 的成员 行(0).getString(1)

和

row(0)(0)

产量

res184: Any = ArrayBuffer([ArrayBuffer(43, 50),online], [ArrayBuffer(51, 61),marketing], [ArrayBuffer(88, 102),growthhacking], [ArrayBuffer(103, 111),入站], [ArrayBuffer(112, 120),启动], [ArrayBuffer(121, 138),内容营销])

然后

row(0)(0)(0)

产量

:28: 错误：Any 不带参数 行(0)(0)(0)

所以，还是卡住了。

5 月 24 日更新：

在尝试使用 .textFile()（非 Spark SQL 方式）并使用本机 Scala JSON 解析功能之后，按照此处的说明：Spark SQL - How to select on dates stored as UTC millis from the epoch?，并遇到此处描述的 Spark json4s 兼容性问题： https://github.com/json4s/json4s/issues/212，如果能解决这些问题，我决定尝试使用 Python 和 pyspark。

5 月 24 日更新 2：

在朋友的帮助下，他建议尝试以下方法：

import scala.collection.mutable.ArrayBuffer
row(0)(0).asInstanceOf[ArrayBuffer[Any]](0).asInstanceOf[ArrayBuffer[Any, String]]

我终于找到了一些进展，因为这行得通：

row(0)(0).asInstanceOf[ArrayBuffer[Any]](0)

并制作：

res53: Any = [ArrayBuffer(43, 50),online]

但是，按照建议进行时：

val a = row(0)(0).asInstanceOf[ArrayBuffer[Any]](0)
a.asInstanceOf[ArrayBuffer[Any, String]]

结果令人沮丧：

22: 错误：类型参数的数量错误 scala.collection.mutable.ArrayBuffer，应该是 1 a.asInstanceOf[ArrayBuffer[Any, String]]

然后试试这个：

a.asInstanceOf[ArrayBuffer[Any]]

产量：

java.lang.ClassCastException: org.apache.spark.sql.catalyst.expressions.GenericRow 不能转换为 scala.collection.mutable.ArrayBuffer

再次，卡住了。

更新 3，5 月 24 日：

所以，在得到朋友的帮助后，我得到了两个可能的解决方案，这两个都不是原始问题的直接答案，但无论如何都要“有点”解决问题。

选项 1（简单的选项，使用 Python）：使用 pyspark - 你可以说：

row[0][0][1]

选项 2（Scala 中丑陋的解决方案）：

val solution = row(0)(0).asInstanceOf[ArrayBuffer[Any]](0).toString().split(" ")(1).split(",")(1).split("]")(0)

产量：

scala> 解决方案 res28: String = 在线

我必须使用 toString() 而不是使用 .asInstanceOf() 的原因是最终的对象包装是：

Any = [ArrayBuffer(43, 50),online]

..我们找不到 .asInstanceOf() 的方法。以下是我们尝试的方法：

row(0)(0).asInstanceOf[ArrayBuffer[Any]](0).asInstanceOf[ArrayBuffer[Any, String]]
row(0)(0).asInstanceOf[ArrayBuffer[Any]](0).asInstanceOf[Row](0).getString(1)
row(0)(0).asInstanceOf[ArrayBuffer[Any]](0).instanceOf[Array[ArrayBuffer[Any], String]]

..但它们都不起作用。

不过，我希望在 Scala 中有一种更优雅的方式来执行此操作，因为 Spark + Scala 的整个“管道构建性质”是吸引我首先使用该软件包的原因..

Answer 1

正如 Twitter 上的回复。这个 Twitter 模式太嵌套了，所以总的来说它非常复杂。但是，我们将来无法为复杂的嵌套字段添加对按名称访问器的支持以简化此操作。

附上 Python 和 Scala 版本。