在Scala中使用RDDS和集合
在Scala中，RDD（Resilient Distributed Datasets）和集合都是用于处理数据的重要概念。
RDD是Spark中的核心数据结构，它代表了一个可分布式、可容错的数据集合。RDD可以通过并行操作进行处理，具有高效的计算能力。RDD的优势包括容错性、可伸缩性和高效性。它可以在内存中缓存数据，以便快速访问。RDD适用于大规模数据处理和分布式计算场景。
集合是Scala中的基本数据结构，用于存储一组元素。Scala提供了多种集合类型，包括列表（List）、集（Set）、映射（Map）等。集合提供了丰富的操作方法，如过滤、映射、排序等，方便对数据进行处理和转换。集合适用于小规模数据处理和单机计算场景。
在使用Scala中的RDD和集合时，可以根据具体的需求选择合适的数据结构。如果需要处理大规模数据或进行分布式计算，可以使用RDD来实现高效的并行计算。如果数据量较小或只需要在单机上进行计算，可以使用集合来简化代码编写。
对于RDD的使用，腾讯云提供了云上数据处理服务Tencent Spark，它基于Spark框架，提供了弹性、高性能的大数据处理能力。您可以通过Tencent Spark来处理和分析大规模数据集，实现数据挖掘、机器学习等应用。
对于集合的使用，腾讯云提供了云函数SCF（Serverless Cloud Function），它是一种无服务器计算服务，可以在云端运行您的代码逻辑。您可以使用SCF来处理小规模数据，实现各种业务逻辑。SCF支持Scala语言，您可以在SCF中使用集合来进行数据处理和计算。
更多关于Tencent Spark的信息，请访问：Tencent Spark产品介绍
更多关于SCF的信息，请访问：腾讯云云函数SCF产品介绍
相关·内容
1回答
我有一个名为index的rdd : RDD(String，String)，我想使用index来处理我的文件。这是代码：
val get = file.map({x =>
val tmp = index.lookup(x).head
tmp
})
问题是我不能在file.map函数中使用索引，我运行了这个程序，它给了我这样的反馈：
14/12/11 16:22:27 WARN TaskSetManager: Lost task 0.0 in stage 3.0 (TID 602, spark2): scala.MatchError: null
org.apache.
浏览 4提问于2014-12-11得票数 4
回答已采纳
1回答
在scala中，有一种方法可以在执行程序中创建本地数据，比如在中的熊猫。在map分区方法中，我希望将迭代器转换为本地dataframe (就像python中的大熊猫dataframe )，这样就可以使用dataframe特性而不是在迭代器上手工编写它们。
浏览 2提问于2018-02-09得票数 2
回答已采纳
尝试为大量RDD文档计算tf-idf分数，每当我尝试将其转换为数据帧时，它总是崩溃。我得到的初始错误是
org.apache.spark.SparkException: Job aborted due to stage failure: Task serialization failed: java.lang.StackOverflowError
然后是这个，重复了很多很多次：
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.Object
浏览 26提问于2019-11-08得票数 0
我尝试使用textFile方法从spark-shell中使用WEB，但出现错误。也许这不是正确的方式。所以谁能告诉我如何从火花上下文访问web URL。
我使用的是spark版本1.3.0；Scala版本2.10.4和Java 1.7.0_21
hduser@ubuntu:~$ spark-shell
Spark assembly has been built with Hive, including Datanucleus jars on classpath
Welcome to
__
/ / / /
\ \/ \/ `/
浏览 0提问于2015-04-20得票数 9
回答已采纳
我对spark和scala语言非常陌生，我希望将列表中的所有RDDs合并如下(List<RDD> to RDD)：
val data = for (item <- paths) yield {
val ad_data_path = item._1
val ad_data = SparkCommon.sc.textFile(ad_data_path).map {
line => {
val ad_data = new AdData(line)
(ad_
浏览 6提问于2015-05-25得票数 8
回答已采纳
在我的代码经过140次迭代之后，我在独立星火集群中得到了以下内存错误。如何在没有内存错误的情况下运行代码？
我有7个具有8GB RAM的节点，其中6GB分配给所有工作人员。主程序也有8GB的RAM。
[error] application - Remote calculator (Actor[akka.tcp://Remote@127.0.0.1:44545/remote/akka.tcp/NotebookServer@127.0.0.1:50778/user/$c/$a#872469007]) has been terminated !!!!!
[info] application - Vi
浏览 3提问于2016-02-08得票数 0
我有一段星火scala代码，如下所示：
val conf = new SparkConf().setAppName("MatrixInversion").setMaster("local")
val sc = new SparkContext(conf)
def main (args: Array[String]) {
var array:Array[Array[Double]]=new Array(2)
for(i<- 0 until 2)
array(i)=new Array(2)
array(0)(0)
浏览 4提问于2015-03-26得票数 4
queueStream函数在StreamingContext中的功能是什么？根据我的理解，它是一个队列，它对传入的DStream进行排队。如果是这样的话，那么如何在具有多个节点的集群中处理它。是否每个节点都有这个queueStream，并且在集群中的所有节点之间对DStream进行分区？在集群设置中，这个queueStream是如何工作的？
我已经阅读了下面的解释在星火流文档)，但我没有完全理解它。请帮我理解一下。
RDDs的队列作为一个流：用于测试带有测试数据的星火流应用程序，还可以使用streamingContext.queueStream(queueOfRDDs)创建一个基于RDDs队
浏览 5提问于2017-08-19得票数 3
1回答
因此，我对Scala很陌生，刚刚开始使用RDDs和functional操作。
我试图迭代我的一对RDD的值，并通过应用已定义的Var1函数来返回Var2中存储的值的平均值，以便最终返回的是一个Var1的唯一列表，每个Var1都有一个单独的AvgVar2。我在弄清楚如何迭代这些值时遇到了很多麻烦。
*编辑:我有以下类型声明：
case class ID: Int, Var1: Int, Var2: Int extends Serializable
我的职能如下：
def foo(rdds: RDD[(ID, Iterable[(Var1, Var2)])]): RDD[(Var1, Avg
浏览 2提问于2019-02-01得票数 0
回答已采纳
3回答
我有一个spark流作业，它每5秒从Kafka读取一次，对传入的数据进行一些转换，然后写入文件系统。
这真的不需要是一个流作业，实际上，我只想每天运行一次，将消息排出到文件系统中。不过，我不确定如何停止这项工作。
如果我向streamingContext.awaitTermination传递一个超时，它不会停止进程，它所做的只是在迭代流时导致进程产生错误(参见下面的错误)。
完成我正在尝试做的事情的最佳方法是什么
这是针对Python上的Spark 1.6的
编辑：
感谢@marios，解决方案是这样的：
ssc.start()
ssc.awaitTermination(10)
ssc.sto
浏览 2提问于2016-01-30得票数 18
回答已采纳
我正在尝试联合到已经分布在我们的集群中的RDDs，并对键进行散列分区。我不需要保留任何顺序，甚至不需要分区，我只希望联合尽可能快。在这个例子中，我实际上想要所有的记录，而不仅仅是不同的记录，而且要保持多重性。
以下是我天真地使用的内容：
val newRDD = tempRDD1.union(tempRDD2)
以下是有人向我推荐的更快的方法，因为它利用了RDDs已经被分区和分布的方式：
val newRDD = tempRDD1.zipPartitions(tempRDD2, preservesPartitioning=true)((iter, iter2) => iter++iter
浏览 0提问于2015-09-10得票数 4
我测试了spark默认PageRank示例，并将迭代设置为1024，然后它抛出stackoverflowerror。我在我的另一个program.How中也遇到了这个问题，我能解决它吗？
object SparkPageRank {
def main(args: Array[String]) {
if (args.length < 3) {
System.err.println("Usage: PageRank <master> <file> <number_of_iterations>")
Syst
浏览 3提问于2014-03-11得票数 8
1回答
我使用Scala2.11.8和Spark2.0.1来运行我的代码。在这一行代码中：
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
object training {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("test").setMaster("local[4]")
val sc = new SparkContext(conf)
浏览 2提问于2016-12-19得票数 0
回答已采纳
我在scala中有以下实用程序函数：
object MyUtiltity {
def processData(data1: org.apache.spark.rdd.RDD[String], data2: org.apache.spark.rdd.RDD[String],
data3: org.apache.spark.rdd.RDD[String]) = {
function1(data1, data3)
function2(data2, data3)
}
private def function1 {...}
private def fun
浏览 0提问于2015-06-05得票数 0
回答已采纳
在为Serialization搜索最佳apache-spark技术的同时，我在下面的链接中发现，使用该框架，在scala中的Serialization将更快、更自动。
由于Scala Pickling具有以下优点。(参考文献- )
因此，我想知道这个Scala Pickling (PickleSerializer)是否可以用于apache-spark而不是KryoSerializer。
如果是的话，需要做什么必要的改变。(示例将有帮助)
如果没有，为什么不呢。(请解释)
提前谢谢。如果我错了请原谅我。
注意:我正在使用scala语言编写apache-spark (版本)。1.4
浏览 4提问于2017-03-21得票数 2
回答已采纳
1回答
我想澄清一下DAG的行为，以及具体如何处理以下工作：
val rdd = sc.parallelize(List(1 to 10).flatMap(x=>x).zipWithIndex,3)
.partitionBy(new HashPartitioner(4))
val rdd1 = sc.parallelize(List(1 to 10).flatMap(x=>x).zipWithIndex,2)
.partitionBy(new HashPartitioner(3))
val rdd2 = rdd.join(rdd1)
rdd2.collect()
这是相关的rdd2.t
浏览 3提问于2017-03-23得票数 2
回答已采纳
1回答
我必须使用Scala/Spark方法和SQL查询来操作RDD。
是否可以通过查询直接使用RDD 操作？
建议的方法(schemaRDD或DataFrame)需要额外的内存泄漏。
经过这样的转换，我记忆中有两个相同的巨大物体。
浏览 2提问于2016-02-08得票数 1
我想将数据保存到MongoDB中，同时从推特上流式传输数据。DStream中的每个RDD都包含带有值的ArrayString，所以我为这些值设置了键，并将它们包装到org.bson.document中。当我尝试将一系列文档写入MongoDB时，我得到了这样一个异常：
ERROR Executor: Exception in task 1.0 in stage 8.0 (TID 9)
java.lang.IllegalArgumentException: clusterListener can not be null
我使用的是Spark MongoDB连接器，所以下面是我的build.sbt文
浏览 2提问于2018-05-27得票数 0
我正在使用spark的scala，并试图结合3000 RDDs。
我正在做一个for循环，迭代3000多个文本文件，创建一个RDD，并将旧的RDD和刚刚从文本文件创建的新RDD联合起来。
这是我的代码：
var tempRDD = sc.textFile(s"" + fileNameArray(ind)).filter(_.split(",",-1).length == cols).filter(line => line.split(",").filter(cell => cell.length > 0).length == c
浏览 2提问于2016-06-16得票数 2
回答已采纳
有一些函数可以删除Spark(SCALA)中的列和行，但是却找不到任何函数来删除整个数据frame.Is，有一种方法可以删除Spark(SCALA)中的数据帧吗？
浏览 0提问于2016-10-15得票数 3
1回答
嗨，我有代码，只是从文档中获取单词计数。在生成输出之前，我还需要使用一个映射来查找数据值。这是密码。
requests
.filter(_.description.exists(_.length > 0))
.flatMap { case request =>
broadcastDataMap.value.get(request.requestId).map {
data =>
val text = Seq(
data.name,
data.taxonym,
浏览 3提问于2016-04-26得票数 4
回答已采纳
如何取消持久化在没有引用的MLlib模型中生成的RDD？
我知道在pyspark中，你可以用sqlContext.clearCache()解压所有的数据帧，除了scala API中的RDDs之外，还有类似的东西吗？此外，有没有一种方法可以只取消一些RDDs的持久化，而不必取消所有RDDs的持久化？
浏览 2提问于2017-02-07得票数 6
回答已采纳
我正在寻找如何在不使用sc.textFile的情况下向foreach传递路径列表(在scala中)。
例子：
myList :Seq[String] = ArrayBuffer (path1, path2, path3)
有没有办法：
var data = sc.textFile(myList)
浏览 2提问于2017-04-24得票数 1
回答已采纳
我在Scala工作，在一台独立的机器(有Windows 10的PC机)上进行星火编程。我是个新手，没有用scala编程的经验。所以我会非常感谢你的帮助。
问题：
我有一个HashMap，hMap1，它的值是整数条目的HashSets (HashMap>)。然后，我将其值(即许多HashSet值)存储在RDD中。代码如下
val rdd1 = sc.parallelize(Seq(hMap1.values()))
现在我有了另一个HashMap，hMap2，同类型的，也就是HashMap>。它的值也存储在RDD中，如
val rdd2 = sc.parallelize(Seq(hMa
浏览 0提问于2017-11-12得票数 0
我正在集群上运行一个小作业，每台机器有15G的内存和8G的磁盘。
作业总是会陷入死锁，最后一条错误消息是：
java.io.IOException: No space left on device
at java.io.FileOutputStream.writeBytes(Native Method)
at java.io.FileOutputStream.write(FileOutputStream.java:345)
at org.apache.spark.storage.DiskBlockObjectWriter$TimeTrackingOutputStream
浏览 1提问于2015-02-11得票数 1
我有许多CSV文件，需要通过文件名的一部分将它们合并到RDD中。
例如，对于以下文件
$ ls
20140101_1.csv 20140101_3.csv 20140201_2.csv 20140301_1.csv
20140301_3.csv 20140101_2.csv 20140201_1.csv 20140201_3.csv
我需要将名为20140101*.csv的文件组合到一个RDD中来处理，等等。
我使用sc.wholeTextFiles读取整个目录，然后根据文件名的模式对文件名进行分组，以形成一个文件名字符串。然后，我将字符串传递给sc.textFile，将文
浏览 3提问于2015-07-21得票数 0
回答已采纳
10回答
定义是：
RDD是不可变的分布式对象集合。
我不太明白这是什么意思。它与存储在硬盘上的数据(分区对象)一样吗?如果是这样，那么为什么RDD可以有用户定义的类(如java、scala或python)？
从这个链接：它提到：
用户以两种方式创建RDDs :通过加载外部数据集，或者通过在其驱动程序中分发对象集合(例如，列表或集合
我对RDD的理解以及与spark和hadoop的关系真的很困惑。
有人能帮帮忙吗。
浏览 91提问于2015-12-23得票数 46
回答已采纳
1回答
我正在测试env上运行火花结构化流。有时，找不到某个检查点文件的作业失败。
其中一个原因可能是卡夫卡的主题保留时间很短。但我已经将.option("failOnDataLoss", "false")添加到SparkSession中了。
我对火花检查点有一些基本的(非常基本的)理解。如果删除检查点dir，我认为应该恢复它。但是，正如我测试的那样，一旦发生了这个错误，删除dir就没有帮助了。我需要使用不同的检查点dir来修复它。
为什么删除检查点dir不起作用？或者是否有一种方法/选项可以帮助避免此错误？
diagnostics: User class t
浏览 0提问于2018-02-02得票数 8
我有一个JavaDStream，它从外部源获取数据。我正在尝试整合星火流和SparkSQL。众所周知，JavaDStream是由JavaRDD组成的。只有当我有一个applySchema()函数时，才能应用JavaRDD ()函数。请帮我把它转换成JavaRDD。我知道scala中有一些函数，它要容易得多。但在爪哇帮我。
浏览 2提问于2014-10-31得票数 4
回答已采纳
我正在研究如何使用spark。
在scala中，我有一段这样的代码：
object GeneMatrixFile {
def main (args: Array[String]) {
val out = new FileWriter("matrix.txt")
val order = 20000
for(i<- 1 to order){
for(j <- 1 to order){
out.write((scala.util.Random.nextDouble()*100).formatted("%3.1f&#
浏览 1提问于2015-04-20得票数 0
1回答
Spark包含一个集合，每个元素代表一个请求。
Scala函数将传递给RDD，对于每个RDD元素，该函数将创建一个修改的请求。
对于每个集合元素\请求，需要引用一个查找表。参考表的最大大小为200行。
如何实现性能和可伸缩性，如何对查找表(在函数中使用)进行建模？
火花广播变量。
分开星火RDD。
Scala不变集合。
也许我还没有考虑过另一种选择。
谢谢
浏览 4提问于2016-01-28得票数 1
回答已采纳
1回答
我是scala和spark的新手，现在我有两个RDD，比如A is (1,2)，(2,3)和B is (4,5)，(5,6)，我想得到RDD，比如(1,2)，(2,3)，(4,5)，(5,6)。但问题是我的数据很大，假设A和B都是10 is。我使用sc.union(A，B)，但它很慢。我在spark UI中看到这个阶段有28308个任务。
有没有更有效的方法来做到这一点？
浏览 3提问于2017-04-22得票数 0
我使用boto3从S3读取文件，这表明比sc.textFile(...)快得多。这些文件大约在300 1GB到1GB之间。这个过程是这样的：
data = sc.parallelize(list_of_files, numSlices=n_partitions) \
.flatMap(read_from_s3_and_split_lines)
events = data.aggregateByKey(...)
在运行此进程时，我会得到以下异常：
15/12/04 10:58:00 WARN TaskSetManager: Lost task 41.3 in stage 0.0 (TI
浏览 12提问于2015-12-04得票数 2
我正在Scala..中用火花GraphX执行Pregel算法。
我的图表包含一百万个节点，它们之间有五百万个边。我的集群非常强大，有几台用于BigData的服务器，每个服务器都有256 My的内存。
我有一个"Java错误“在洗牌阶段，在执行过程中，经过超过20分钟的处理：任务丢失了.我将分析这些参数：
I加载图的方式及其持久性(StorageLevel)分析
内存的分析：节点、边缘和发送的消息
分析火花提交命令时星火中设置的参数：执行器的数量，每个执行器的，serialization...
浏览 1提问于2016-05-01得票数 0
我正在尝试用MLLib在Pyspark中运行逻辑回归。模型运行了，但我无法获得任何指标。
我的数据是csv格式，我按如下方式进行转换：
def load(prefix):
lines = spark.read.text(prefix).rdd
parts = lines.map(lambda row: row.value.split(","))
ratingsRDD = parts.map(lambda p: Row(pct = str(p[0]), date = str(p[1]), res_burg_label=int(p[2]), com_burg
浏览 1提问于2018-03-17得票数 0
我最近开始在纱线上使用spark，在调优我的程序时发现了一个问题。
当SparkContext被初始化为sc并准备从hdfs读取文本文件时，将调用textFile(path, defaultMinPartitions)方法。我跟踪了spark源代码中的第二个参数，最终找到了这个参数：CoarseGrainedSchedulerBackend.scala中的conf.getInt("spark.default.parallelism", math.max(totalCoreCount.get(), 2))
我没有指定属性“spark.default.parallelism” an
浏览 2提问于2014-07-22得票数 1
3回答
我有一对RDD，格式为：RDD[(String, String)]和一个文件中的密钥列表。我想要一个RDD，它只包含那些与列表中的键匹配的键-值对。我如何实现这一点(最好是在Scala中)？谢谢。
浏览 0提问于2017-01-12得票数 1
回答已采纳
我是Spark和Scala的新手，但有以下UseCase可以使用。
我有n条tweet和m家公司，n >> m，都在一个RDD中。我想把他们加在一起，以便检查推文中有哪些公司。
在迭代编程中，我会做一些像嵌套循环连接这样的事情，并使用自定义匹配函数。
在Scala和Spark中，我在RDD上使用笛卡尔函数，但这给了我一个内存不足的异常。我已经将spark.executor.memory设置为12g
以下是我的代码
val tweets = sc.textFile("/Users/romeokienzler/Documents/proj/conferences/velocit
浏览 2提问于2015-10-28得票数 1
我有一个csv文件，其中包含像Z1:A，Z2:B等用逗号分隔的值。我想: 1.创建键值，其中包含每个键的频率数(这部分已经完成了)。2.重写我的数组(或者元组？)通过保持Zx (其中x可以是任意整数)并丢弃冒号和所有跟随(:A )的内容。
是我的虚拟文件(为了简单起见，只有Z1和Z2)。
所以我把我的cvs装进了val
val example1 = sc.textFile("/Users/....../Spark_stack/example_1.csv")
然后，我执行map-还原，以获得我想要的结果。
val counts = example1.flatMap(line =&
浏览 11提问于2017-04-11得票数 0
回答已采纳
3回答
我一直在使用Scala + Spark和教程。
在获得我的预测后，我需要每个用户的前3个项目。
val predictions =
model.predict(usersProducts).map { case Rating(user, product, rate) =>
(user, product, rate)
}
我试过了：
def myPrint(x:(Int, Int, Double)) = println(x._1 + ":" + x._2 + " - " +x._3)
predictions.collect().sortBy(-
浏览 0提问于2014-11-11得票数 0
1回答
我有一个关于同时加入多个rdds的问题。我有大约8对RDD的数据类型:RDD [(String，mutable.HashSetString)]。我想按键加入他们。我可以使用spark的join或cogroup加入2吗？但是，有没有内置的方法可以做到这一点呢？我可以一次加入两个，然后将结果rdd与下一个rdd结合起来，但是如果有更好的方法，我想使用它。
浏览 1提问于2015-08-31得票数 2
1回答
在PySpark中创建类的
、、、、
我必须将Scala代码转换为python。
scala代码将字符串的RDD转换为case类的RDD。守则如下：
case class Stock(
stockName: String,
dt: String,
openPrice: Double,
highPrice: Double,
lowPrice: Double,
closePrice: Double,
浏览 1提问于2020-06-03得票数 1
回答已采纳
我有一个Scala程序，它使用REST逐批获取数据，一旦检索到所有数据，我就对它们进行操作。
当前程序：
对于每个批处理，创建RDD并将其与使用前一个API调用rdd.union(currentRdd)创建的前一个RDD合并。
在最终RDD上操作
一个简单的程序来重现这个问题：
def main(args: Array[String]) = {
val conf = new SparkConf().setAppName("Union test").setMaster("local[1]")
val sc = new S
浏览 2提问于2019-07-22得票数 3
回答已采纳
我试着运行一个示例星火流代码。但我知道这个错误：
16/06/02 15:25:42 ERROR streaming.StreamingContext: Error starting the context, marking it as stopped
java.lang.IllegalArgumentException: requirement failed: No output operations registered, so nothing to execute
at scala.Predef$.require(Predef.scala:233)
at org.apache.spark
浏览 3提问于2016-06-03得票数 1
/Dataset执行引擎有几个非常高效的时间和空间优化(例如，InternalRow和expression )。根据许多文档，对于大多数分布式算法，它似乎是一个比RDD更好的选择。
然而，我做了一些源代码研究，但仍然不相信。毫无疑问，InternalRow更紧凑，可以节省大量内存。但是，算法的执行可能不会更快地保存预定义表达式。也就是说，在org.apache.spark.sql.catalyst.expressions.ScalaUDF的源代码中，每个用户定义的函数都做3件事：
将催化剂类型(用于InternalRow)转换为scala类型(用于GenericRow)。
应用函数
浏览 2提问于2016-05-30得票数 10
--我尝试从Socket获取数据以附加到dataframe --我接收了数据并将其执行到Seq()中，但是当我使用forEach将它们附加到dataframe时，有一个问题--这是我的代码：
object CustomReceiver {
def main(args: Array[String]): Unit = {
StreamingExamples.setStreamingLogLevels()
// Create the context with a 1 second batch size
val spark: SparkSession = SparkS
浏览 2提问于2021-06-21得票数 0
回答已采纳
我正在处理一个初始大小为569 MB的数据集，计算TF-IDF度量。尽管我最终得到了结果，但我仍然得到以下错误： WARN scheduler.TaskSetManager: Lost task 13.0 in stage 11.0 (TID 84, X.X.X.X, executor 0): FetchFailed(null, shuffleId=4, mapId=-1, reduceId=4, message=
org.apache.spark.shuffle.MetadataFetchFailedException: Missing an output location for shu
浏览 33提问于2021-10-31得票数 1
我正在使用Spark，并创建了两个idential长度数组，一个是tweet的时间，另一个是tweet的文本。我希望将它们合并成一个数据结构(也许是一个元组？)我可以过滤的时间和文字的推特，但我正在挣扎后，结合起来，如何执行。
scala> val split_time = split_date.map(line => line.split(":")).map(word =>
(word(0)))
split_time: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[28] at map
at <
浏览 0提问于2017-12-08得票数 1
回答已采纳
我有一个RDD[MapString，Any]，我正在尝试将它转换为Dataframe。我没有可以指定Dataframe的架构。
我试着做了一个rdd.toDF，但是没有帮助。它出现了一个错误，如下所示。
Exception in thread "main" java.lang.ClassNotFoundException: scala.Any
at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
at java.lang.ClassLoader.loadClass(ClassLoader.ja
浏览 2提问于2019-11-14得票数 2
2回答
我试图从Spark2.1作业中读取Cassandra表(mykeyspace.mytable) (使用Scala2.11)：
val myDataset = sqlContext
.read
.format("org.apache.spark.sql.cassandra")
.options(Map("table" -> "mytable", "keyspace" -> "mykeyspace"))
.load()
myDataset.show()
print
浏览 3提问于2017-07-31得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
