如何将RDD [Row]转换为RDD [String](How to convert RDD[Row] to RDD[String])

我有一个名为source的DataFrame，一个来自mysql的表

val source = sqlContext.read.jdbc(jdbcUrl, "source", connectionProperties)

我已将它转换为rdd

val sourceRdd = source.rdd

但是它的RDD [Row]我需要RDD [String]来进行转换

source.map(rec => (rec.split(",")(0).toInt, rec)), .subtractByKey(), etc..

谢谢

I have a DataFrame called source, a table from mysql

val source = sqlContext.read.jdbc(jdbcUrl, "source", connectionProperties)

I have converted it to rdd by

val sourceRdd = source.rdd

but its RDD[Row] I need RDD[String] to do transformations like

source.map(rec => (rec.split(",")(0).toInt, rec)), .subtractByKey(), etc..

Thank you

原文：https://stackoverflow.com/questions/44067476

更新时间：2019-06-05 17:32

最满意答案

你可以使用Row. mkString(sep: String): String Row. mkString(sep: String): String map调用中的Row. mkString(sep: String): String方法，如下所示：

val sourceRdd = source.rdd.map(_.mkString(","))

您可以根据需要更改","参数。

希望对您有所帮助，最诚挚的问候。

You can use Row. mkString(sep: String): String method in a map call like this :

val sourceRdd = source.rdd.map(_.mkString(","))

You can change the "," parameter by whatever you want.

Hope this help you, Best Regards.

2017-05-19

相关问答

你可以使用Row. mkString(sep: String): String Row. mkString(sep: String): String map调用中的Row. mkString(sep: String): String方法，如下所示： val sourceRdd = source.rdd.map(_.mkString(","))

您可以根据需要更改","参数。 希望对您有所帮助，最诚挚的问候。 You can use Row. mkString(sep: String): Stri

...

PySpark Row只是一个tuple ，可用于此。 所有你需要的是一个简单的map (或flatMap如果你想要平整行)以及list ： data.map(list)

或者如果你期望不同的类型： data.map(lambda row: [str(c) for c in row])

PySpark Row is just a tuple and can be used as such. All you need here is a simple map (or flatMap if you

...

这里至少有几个问题： 不，你真的不能将一行转换为Vector：一行是Spark SQL理解的潜在不同类型的集合。 Vector不是本机spark sql类型 SQL语句的内容与您尝试使用KMeans实现的内容之间似乎不匹配：SQL正在执行聚合。 但是KMeans期望一系列单独的数据点形式为Vector(它封装了一个Array[Double] )。 那么 - 为什么要为KMeans操作提供sum和average ？ 仅在这里解决＃1：你将需要做一些事情： val doubVals =

...

Cloudera＆Spark 2.0？ 嗯，没想到我们支持了:) 无论如何，首先你不需要在RDD上调用.persist() ，这样你就可以删除那个位。 其次，由于Person是一个案例类，你应该将其名称大写。 最后，在Spark 2.0中，您不再调用import sqlContext.implicits._来隐式构建DataFrame模式，现在调用import spark.implicits._ 。 您的错误消息暗示了这一点。 There was a simple mistake where I

...

正如你可以看到数据类型的差异 第一个是RDD[(String, Int)]这是一个包含(String, Int)的Tuple2的RDD ，因此您可以以val prod = x._1访问此值，第一个值为String ，第二个Integer值为x._2 。 既然它是一个元组，你不能访问为val prod = x(0) 第二个是可以访问val prod = x.getString(0)或val prod = x(0) RDD[org.apache.spark.sql.Row] 我希望这有助于！ As y

...

您可以使用sqlContext从Cassandra表中读取数据，它将返回一个DataFrame，并且当您使用sparkContext读取文本文件时，它将返回RDD，然后您可以将其转换为DataFrame。 如果您的文本文件是CSV，则Spark 2.0支持csv数据源，它会通过deafult返回一个DataFrame。 请看这.. https://spark.apache.org/releases/spark-release-2-0-0.html#new-features和https://gith

...

你不能有RDD[String, String] 。 RDD只接受1个type parameter所以你想要的是RDD[(String, String)] 。 df.rdd

.map(row => {

val id = row.getString(0)

val country = row.getString(1)

val title = row.getString(2)

val jsonString = s"{country: $country, title:

...

这是一个小例子，它导致了同样的问题(我省略了Date ，将其替换为String ，这不是重点)： val myRdd = sc.makeRDD(List(

List[AnyRef]("date 1", "blah2", (11: java.lang.Integer), "baz1"),

List[AnyRef]("date 2", "blah3", (5: java.lang.Integer), "baz2"),

List[AnyRef]("date 3", "blah4", (1:

...

考虑到输入文件的每一行都是一个以逗号分隔的字符串表示的假设向量，这是一个非常基本的操作。 您只需要映射每个字符串entrie，将其拆分在分隔符上，然后从中创建一个密集的Vector： val parsedData = rawData.map(s => Vectors.dense(s.split(',').map(_.toDouble)))

This is a very basic operation considering that each line of your input file is

...

尝试： sqlContext.read().parquet(path).as(encoder).rdd().toJavaRDD();

Try: sqlContext.read().parquet(path).as(encoder).rdd().toJavaRDD();

---

1.本站遵循行业规范，任何转载的稿件都会明确标注作者和来源；2.本站的原创文章，会注明原创字样,如未注明都非原创，如有侵权请联系删除！；3.作者投稿可能会经我们编辑修改或补充；4.本站不提供任何储存功能只提供收集或者投稿人的网盘链接。