如何在Spark结构化流媒体联接中选择最新记录
我正在使用spark-sql 2.4.x版本，对于Cassandra-3.x版本使用datastax-spark-cassandra-connector.连同卡夫卡. 我有如下货币样本的汇率元数据: val ratesMetaDataDf = Seq( ("EUR","5/10/2019","1.130657","USD"), ("EUR","5/9/2019","
..
apache-spark-dataset相关内容
Spark CSV-找不到适用于实际参数的适用构造函数/方法
在java spark应用程序中的类型数据集的过滤器和映射上使用lambda函数时，我遇到了一个问题. 我遇到此运行时错误 ERROR CodeGenerator: failed to compile: org.codehaus.commons.compiler.CompileException: File 'generated.java', Line 130, Column 126:
..
Spark:如果DataFrame具有架构，DataFrame如何成为Dataset [Row]
此博客文章显示DataFrame具有架构. 以博客文章中将RDD转换为DataFrame的示例为例:如果DataFrame与Dataset[Row]相同，则将RDD转换为DataFrame应该很简单 val rddToDF = rdd.map(value => Row(value)) 但是相反，它表明就是这个 val rddStringToRowRDD = rdd.map(va
..
如何使用类型化的数据集将多值列拆分为单独的行?
我面临一个如何将多值列(即List[String])拆分为单独的行的问题. 初始数据集具有以下类型:Dataset[(Integer, String, Double, scala.List[String])] +---+--------------------+-------+--------------------+ | id| text | value
..
Spark数据帧-按键减少
假设我有一个这样的数据结构，其中ts是一些时间戳记 case class Record(ts: Long, id: Int, value: Int) 给出大量的这些记录，我想以每个ID的时间戳都最高的记录结束.我认为使用RDD API可以完成以下代码: def findLatest(records: RDD[Record])(implicit spark: SparkSession
..
Spark Dataset API-加入
我正在尝试使用Spark 数据集 API，但在进行简单连接时遇到一些问题. 假设我有两个带有字段的数据集:date | value，那么在DataFrame的情况下，我的联接如下所示: val dfA : DataFrame val dfB : DataFrame dfA.join(dfB, dfB("date") === dfA("date") ) 但是对于Dataset，有.
..
分区数据以有效联接Spark数据帧/数据集
我需要根据某些共享键列将许多DataFrame一起join.对于键值RDD，可以指定一个分区程序，以便将具有相同键的数据点改组到同一执行程序，从而提高连接效率(如果在join之前进行了与改组相关的操作).可以在Spark DataFrames或DataSet上做同样的事情吗? 解决方案 如果您知道要多次加入它，则可以在装入数据框后repartition val users = spark
..
Spark数据集聚合类似于RDD聚合(零)(累加，组合器)
RDD有一个非常有用的方法聚合，它允许累加一些零值并将其合并到各个分区中.有没有办法用Dataset[T]做到这一点.就我通过Scala doc看到的规范而言，实际上没有能力做到这一点.甚至reduce方法也只允许对以T为两个参数的二进制运算执行操作.有什么原因吗?如果有什么功能可以做到这一点? 非常感谢！ VK 解决方案 可以使用两种不同的类在Dataset API中实现类似
..
Spark动态DAG与硬编码DAG相比要慢得多并且有所不同
我在spark中进行了一项操作，该操作应针对数据帧中的几列执行.通常，有两种可能性可以指定此类操作 硬编码 handleBias("bar", df) .join(handleBias("baz", df), df.columns) .drop(columnsToDrop: _*).show 从名称列表动态生成它们 var isFirst = true var re
..
数据框到类型为Any的数据集
我最近从Spark 1.6迁移到Spark 2.X，并且我想尽可能地从Dataframes迁移到Datasets.我尝试过这样的代码 case class MyClass(a : Any, ...) val df = ... df.map(x => MyClass(x.get(0), ...)) 如您所见，MyClass具有一个类型为Any的字段，因为在编译时我不知道我使用x.get
..
Spark 2.0隐式编码器，当类型为Option [Seq [String]](标量)时处理缺少的列
当我们的数据源中缺少某些类型为Option [Seq [String]]的列时，我在编码数据时遇到了一些麻烦.理想情况下，我希望丢失的列数据用None填充. 场景: 我们正在读取的某些实木复合地板文件中有 column1 ，但没有 column2 . 我们将这些镶木文件中的数据加载到Dataset中，并将其转换为MyType. case class MyType(colum
..
Spark数据集:示例:无法生成编码器问题
引发世界并尝试使用我在网上找到的用scala编写的数据集示例 通过SBT运行它时，我不断遇到以下错误 org.apache.spark.sql.AnalysisException: Unable to generate an encoder for inner class 任何想法我都忽略了什么 还可以随时指出编写相同数据集示例的更好方法 谢谢 > sbt>
..
什么时候使用Spark DataFrame/Dataset API，什么时候使用普通的RDD?
Spark SQL DataFrame/Dataset执行引擎具有几个非常高效的时间&空间优化(例如InternalRow& expression codeGen).根据许多文档，对于大多数分布式算法而言，它似乎比RDD更好. 但是，我进行了一些源代码研究，但仍然没有被说服.我毫不怀疑InternalRow更加紧凑，可以节省大量内存.但是算法的执行可能不会更快地保存预定义的表达式.即，在or
..
为什么出现错误“无法找到存储在数据集中的类型的编码器"?使用案例类编码JSON时?
我写了火花工作: object SimpleApp { def main(args: Array[String]) { val conf = new SparkConf().setAppName("Simple Application").setMaster("local") val sc = new SparkContext(conf) val ctx = ne
..
Spark SQL中的数组交集
我有一个表，该表的数组类型列名为writer，其值类似于array[value1, value2]，array[value2, value3] ...等. 我正在执行self join以获得数组之间具有公共值的结果.我试过了: sqlContext.sql("SELECT R2.writer FROM table R1 JOIN table R2 ON R1.id != R2.id W
..
为什么在Apache Spark SQL中列更改为可为空?
即使在DataFrame中没有NaN值，执行某些功能后为什么还要使用nullable = true. val myDf = Seq((2,"A"),(2,"B"),(1,"C")) .toDF("foo","bar") .withColumn("foo", 'foo.cast("Int")) myDf.withColumn("foo_2", when($
..
如何按组使用roximateQuantile?
Spark具有SQL函数percentile_approx()，它的Scala副本是df.stat.approxQuantile(). 但是，Scala副本不能用于分组数据集，例如df.groupby("foo").stat.approxQuantile()，如下所示: https://stackoverflow.com/a/51933027 . 但是可以用SQL语法进行分组和百分位.
..
如何从SparkSQL DataFrame中的MapType列获取键和值
我在一个实木复合地板文件中有数据，该文件包含2个字段:object_id: String和alpha: Map. 将其读入sparkSQL的数据帧中，其模式如下所示: scala> alphaDF.printSchema() root |-- object_id: string (nullable = true) |-- ALPHA: map (nullable = true)
..
火花createOrReplaceTempView与createGlobalTempView
Spark Dataset 2.0提供了两个功能createOrReplaceTempView和createGlobalTempView.我无法理解这两个功能之间的基本区别. 根据 API文档: createOrReplaceTempView :此操作的生命周期 临时视图与用于创建此数据集的[[SparkSession]]绑定在一起. 因此，当我调用sparkSession.clos
..
如何创建地图数据集?
我正在使用Spark 2.2，尝试在Map的Seq上调用spark.createDataset时遇到麻烦. Spark Shell会话中的代码和输出如下: // createDataSet on Seq[T] where T = Int works scala> spark.createDataset(Seq(1, 2, 3)).collect res0: Array[Int] = A
..