scala – 我们应该并行化DataFrame,就像我们在训练之前并行化Seq

发布时间：2020-12-16 18:41:57 所属栏目：安全来源：网络整理

导读：考虑一下这里给出的代码, https://spark.apache.org/docs/1.2.0/ml-guide.html import org.apache.spark.ml.classification.LogisticRegressionval training = sparkContext.parallelize(Seq( LabeledPoint(1.0,Vectors.dense(0.0,1.1,0.1)),LabeledPoint(0.

考虑一下这里给出的代码,

https://spark.apache.org/docs/1.2.0/ml-guide.html

import org.apache.spark.ml.classification.LogisticRegression
val training = sparkContext.parallelize(Seq(
  LabeledPoint(1.0,Vectors.dense(0.0,1.1,0.1)),LabeledPoint(0.0,Vectors.dense(2.0,1.0,-1.0)),1.3,1.0)),LabeledPoint(1.0,1.2,-0.5))))

val lr = new LogisticRegression()
lr.setMaxIter(10).setRegParam(0.01)

val model1 = lr.fit(training)

假设我们使用sqlContext.read()将“training”作为数据框读取,应该
我们还是做点什么的

val model1 = lr.fit(sparkContext.parallelize(training)) // or some variation of this

或者,在传递dataFrame时,fit函数将自动处理并行计算/数据

问候,

解决方法

DataFrame是一种分布式数据结构.并行化既不需要也不可能. SparkConext.parallelize方法仅用于驻留在驱动程序内存中的分布式本地数据结构.您不应该习惯于分发大型数据集,更不用说重新分发RDD或更高级别的数据结构(就像您在上一个问题中所做的那样)

sc.parallelize(trainingData.collect())

如果要在RDD / Dataframe(数据集)之间进行转换,请使用旨在执行此操作的方法：

>从DataFrame到RDD：

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.Row
import org.apache.spark.rdd.RDD

val df: DataFrame  = Seq(("foo",1),("bar",2)).toDF("k","v")
val rdd: RDD[Row] = df.rdd

>将RDD格式化为DataFrame：

val rdd: RDD[(String,Int)] = sc.parallelize(Seq(("foo",2)))
val df1: DataFrame = rdd.toDF
// or
val df2: DataFrame = spark.createDataFrame(rdd) // From 1.x use sqlContext

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!