Блог

У меня есть очень простая таблица базы данных cassandra endless_loop в пространстве test ключей . Таблица выглядит так:

 id
1
2
 

Почему я получаю бесконечный цикл с этим кодом, который объединяет sort -, filter — и union -операции?

 import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.cassandra._

object sparkUnderstandEndlessLoops {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder
      .appName("Endless Loop")
      .config("spark.master", "local")
      .config("spark.cassandra.connection.host", "localhost")
      .config("spark.cassandra.connection.port", "9042")
      .getOrCreate()
    val sc = spark.sparkContext

    val df = spark.read.cassandraFormat("endless_loop", "test").load()
    val df1 = df.sort("id")
    val df2 = df.filter(_=>false)
    val df3 = df1.union(df2)

    df3.show()
  }
}

 

Я использую следующий sbt файл:

 scalaVersion := "2.12.11"
    libraryDependencies  = "org.apache.spark" %% "spark-core" % "2.4.5"
    libraryDependencies  = "org.apache.spark" %% "spark-sql" % "2.4.5"
    libraryDependencies  = ("org.apache.cassandra" % "cassandra-all" % "3.11.6")
                              .exclude("net.jpountz.lz4", "lz4")
                              .exclude("ch.qos.logback", "logback-classic")
                              .exclude("ch.qos.logback", "logback-core")
                              .exclude("org.slf4j", "log4j-over-slf4j")
    libraryDependencies  = "com.datastax.spark" %% "spark-cassandra-connector" % "2.4.3"
 

Редактировать:

Та же ошибка по-прежнему возникает, когда последние строки изменяются на

 val df = spark.read.cassandraFormat("endless_loop", "test").load()
val df1 = df.sort("id")
val df2 = df1.union(df)

df2.show()