Использование определенной функции Spark 2.4? - PullRequest
Новая
       28

Использование определенной функции Spark 2.4?

1 голос
/ 08 апреля 2019

Я использую алгоритм kmeans, я создаю VectorAssembler, устанавливаю inputcols в ("долгота", "широта") и outputCol в ("местоположение"). Мне нужно кластеризовать мои данные из файла json в 3 кластера. Я классифицирую данные по долготе и широте и создаю местоположение вектора, чтобы соединить оба. Расположение и широта являются DoubleType. Я думаю, что это из-за вектора местоположения Я получаю сообщение об ошибке ниже: 

19/04/08 15:20:56 ERROR Executor: Exception in task 0.0 in stage 1.0 (TID 1)
org.apache.spark.SparkException: Failed to execute user defined function(VectorAssembler$$Lambda$1629/684426930: (struct<latitude:double,longitude:double>) => struct<type:tinyint,size:int,indices:array<int>,values:array<double>>)
    at org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage1.processNext(Unknown Source)

вот мой код: 

import org.apache.spark.sql.{SQLContext, SparkSession}
import org.apache.spark.sql.types.{DataTypes, DoubleType, StructType}
import org.apache.spark.ml.clustering.{KMeans, KMeansModel}
import org.apache.spark.ml.evaluation.ClusteringEvaluator
import org.apache.spark.ml.feature
import org.apache.spark.{SparkConf, SparkContext, sql}
import org.apache.spark.ml.feature.{Binarizer, Interaction, VectorAssembler}
import org.apache.spark.sql
import org.apache.spark.sql.expressions.UserDefinedFunction
//plotting




object Clustering_kmeans {

  def main(args: Array[String]): Unit = {

    println("hello world me")

    // Spark Session
   val  sc = SparkSession.builder().appName("Clustering_Kmeans").master("local[*]").getOrCreate()


    import sc.implicits._
    sc.sparkContext.setLogLevel("WARN")
    // Loads data.
    val stations = sc.sqlContext.read.option("multiline",true).json("/home/aymenstien/Téléchargements/Brisbane_CityBike.json")
// trans
    val st = stations.withColumn("longitude", $"longitude".cast(sql.types.DoubleType))
      .withColumn("latitude", $"latitude".cast(sql.types.DoubleType)).cache()


    val stationVA = new VectorAssembler().setInputCols(Array("latitude","longitude")).setOutputCol("location")
    val stationWithLoc =stationVA.transform(st)


//print

    stationWithLoc.show(truncate = false)
    stationWithLoc.printSchema()
    //val x = st.select('longitude).as[Double].collect()
    //val y = st.select('latitude).as[Double].collect()

//st.printSchema()

  }

}

вот Схема 

root
 |-- address: string (nullable = true)
 |-- coordinates: struct (nullable = true)
 |    |-- latitude: double (nullable = true)
 |    |-- longitude: double (nullable = true)
 |-- id: double (nullable = true)
 |-- latitude: double (nullable = true)
 |-- longitude: double (nullable = true)
 |-- name: string (nullable = true)
 |-- position: string (nullable = true)
 |-- location: vector (nullable = true)

1 Ответ

0 голосов
/ 08 апреля 2019

Поскольку у вас есть nullable = true для всех функций, если у вас есть нулевые значения, VectorAssembler выдаст ошибку. Попробуйте установить handleInvalid на "skip". Это отфильтрует строки с любыми нулями. 

val stationVA = new VectorAssembler().
                     setInputCols(Array("latitude","longitude")).
                     setOutputCol("location").
                     setHandleInvalid("skip")
...