带有泛型Dataset[T]参数的Scala Spark函数，也返回Dataset[T]？
带有泛型Dataset[T]参数的Scala Spark函数是一种能够处理数据集的函数，其中的泛型T表示数据集中的元素类型。该函数接受一个Dataset[T]作为输入参数，并且也返回一个Dataset[T]作为输出结果。
泛型Dataset[T]参数的函数在Spark中非常常见，它们可以用于对数据集进行各种转换、过滤、聚合等操作。通过使用泛型参数T，这些函数可以适用于不同类型的数据集，提供了更大的灵活性和通用性。
优势：
- 通用性：泛型Dataset[T]参数的函数可以适用于不同类型的数据集，无需针对每种数据类型编写特定的函数，提高了代码的复用性和可维护性。
- 类型安全：使用泛型参数T可以在编译时进行类型检查，避免了在运行时出现类型错误的可能性。
- 扩展性：通过使用泛型参数T，函数可以适应未来可能出现的新数据类型，无需修改现有代码。
应用场景：
- 数据转换：可以使用泛型Dataset[T]参数的函数对数据集进行各种转换操作，如映射、过滤、排序等。
- 数据聚合：可以使用泛型Dataset[T]参数的函数对数据集进行聚合操作，如求和、计数、平均值等。
- 数据分析：可以使用泛型Dataset[T]参数的函数进行数据分析，如统计、分类、预测等。
推荐的腾讯云相关产品： 腾讯云提供了一系列与Spark相关的产品和服务，可以帮助用户更好地使用和管理Spark集群，如下所示：
- 腾讯云EMR：腾讯云弹性MapReduce（EMR）是一种大数据处理平台，支持Spark等多种计算框架，可以快速搭建和管理Spark集群。 产品链接：https://cloud.tencent.com/product/emr
- 腾讯云COS：腾讯云对象存储（COS）是一种高可用、高可靠、低成本的云端存储服务，可以用于存储和管理Spark计算过程中产生的大量数据。 产品链接：https://cloud.tencent.com/product/cos
- 腾讯云SCF：腾讯云云函数（SCF）是一种事件驱动的无服务器计算服务，可以用于执行Spark函数，实现按需计算和资源弹性扩展。 产品链接：https://cloud.tencent.com/product/scf
请注意，以上推荐的产品仅为腾讯云提供的一些相关服务，其他云计算品牌商也提供类似的产品和服务。
相关·内容
我是scala的新手，所以很难想象在下面的代码中发生了什么。
我有一个Dataset，它有一个映射函数。
这是映射函数
def map[U](func: MapFunction[T, U], encoder: Encoder[U]): Dataset[U]
正如我们所看到的，它包含两个参数，func trait (interface in java) and Encoder trait
在我的代码中，有一个函数在下面使用我的数据集，如下所示
def myDef(spark : SparkSession, dataset: DataSet) : Dataset[MyType] {
浏览 3提问于2022-08-06得票数 0
回答已采纳
当我尝试用Scala参数化类型重写java泛型函数名时，我得到了错误-方法读取有不兼容的类型。
这是我试图覆盖的java抽象函数。
public abstract class SRType{
//Other abstract functions
public abstract <T> T read()throws IOException;
}
我在scala中以如下方式使用这个抽象类：
abstract class SRType(val name: String) extends org.xyz.SRType {
// to convert to
浏览 11提问于2018-01-11得票数 1
回答已采纳
我是Scala的新手&我试着理解Spark的。“flatMap”函数有这样的文档
def
flatMap[U](f: (T) ⇒ TraversableOnce[U])(implicit arg0: ClassTag[U]): RDD[U]
Return a new RDD by first applying a function to all elements of this RDD, and then flattening the results.
虽然我知道flatMap到底是做什么的，但是理解文档似乎太神秘了(比如U、f、T等字母)。如果有人能解释这个文档的每个部分到底传达了什
浏览 2提问于2015-06-05得票数 1
回答已采纳
1回答
我有两个数据集：
implicit val spark: SparkSession = SparkSession
.builder()
.appName("app").master("local[1]")
.config("spark.executor.memory", "1g")
.getOrCreate()
import spark.implicits._
val ds1 = /*read csv file*/.as[caseClass1]
val ds2 = /*read csv file*/.as
浏览 3提问于2017-11-05得票数 0
回答已采纳
我想解决这个问题，但无法进一步解决。有人能帮帮忙吗
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat
class KeyBasedOutput[T >: Null, V <: AnyRef] extends MultipleTextOutputFormat[T , V] {
override def generateFileNameForKeyValue(key: T, value: V, leaf: String) = {
key.toString
}
override def generateActual
浏览 0提问于2014-09-23得票数 3
我在使用spark_apply的Spark中遇到了GC开销超过限制错误。以下是我的规格：
sparklyr v0.6.2 Spark v2.1.0 4个工作进程，具有8个内核和29G内存
closure get_dates一次从Cassandra拉取一行数据。总共大约有200k行。这个过程运行了大约一个半小时，然后出现了这个内存错误。
我已经用spark.driver.memory做了实验，它应该会增加堆的大小，但它不起作用。
有什么想法吗？下面的用法
> config <- spark_config()
> config$spark.executor.cores = 1 #
浏览 3提问于2018-03-10得票数 1
1回答
我定义了一个带有Enter和Exit类型的类型InterfaceExecuter，两者都是struct。在另一个类中，我希望创建一个方法来实例化实现我的接口的类，如下所示：
public T<TEntry, TExit> Initiate<T<TEntry, TExit>>()
where T: InterfaceExecuter<TEntry, TExit>, new()
where TEntry:struct
where TExit:struct
{
return ne
浏览 1提问于2017-10-24得票数 3
回答已采纳
嗨，我正在试图找到评级直方图，使用scastie program...here的实现。
星条旗的设置
scalacOptions ++= Seq(
"-deprecation",
"-encoding", "UTF-8",
"-feature",
"-unchecked"
)
libraryDependencies ++= Seq(
"org.
浏览 4提问于2020-09-10得票数 0
回答已采纳
1回答
我只是尝试在Flink中使用Scala类型的类。我定义了以下类型的类接口：
trait LikeEvent[T] {
def timestamp(payload: T): Int
}
现在，我想考虑一下这样的DataSet of LikeEvent[_]：
// existing classes that need to be adapted/normalized (without touching them)
case class Log(ts: Int, severity: Int, message: String)
case class Metric(ts: Int, name:
浏览 5提问于2020-04-13得票数 2
回答已采纳
2回答
我试着这样定义一个简单的特征：
scala> trait T { def p[A,B]: Map[A,B] }
defined trait T
scala> new T { def p = Map(0 -> 1) }
<console>:7: error: object creation impossible, since method p in trait T of type [A,B]Map[A,B] is not defined
new T { def p = Map(0 -> 1) }
^
怎么会这样?
谢谢
浏览 0提问于2010-07-21得票数 2
回答已采纳
1回答
我正在尝试从tiff图像中获取海量高程数据，我有一个csv文件。csv文件内容也有纬度、经度等属性。循环通过csv文件，获取纬度和经度，并调用elevation方法，代码如下。参考
package main.scala.sample
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql._
import org.apache.spark.sql.functions._
import org.locationtech.rasterframes._
import o
浏览 6提问于2021-07-15得票数 0
3回答
我只是想知道Scala泛型的实现细节。在C#中，可以将类声明为：
class Foo<T1>{}
class Foo<T1, T2>{}
然而，在Scala中，同样的事情必须声明为
class Foo0[T1]{}
class Foo1[T1, T2]{}
注意多个泛型参数的类名是如何被强制更改的。有没有什么原因让Scala决定走这条路而不是我觉得更优雅的C#呢？我知道这可能是一个相当小的吹毛求疵，但我很好奇其中的原因。
浏览 2提问于2009-06-15得票数 8
回答已采纳
下面的代码将导致NullPointerException。我不确定，这个异常是发生在某些行上，还是总是因为dataframe很大而无法指向行。
def removeUnwantedLetters(str: String): String = {
str.split("\\W+").filter(word => (word.matches("[a-z]+") && (word.length > 1))).mkString(" ")
}
val myudf = spark.udf.register("le
浏览 0提问于2018-11-30得票数 0
回答已采纳
我正在对抛出SparkSession的Object is not an instance of declaring class执行查询，下面是下面的代码
Dataset<Row> results = spark.sql("SELECT t1.someCol FROM table1 t1 join table2 t2 on t1.someCol=t2.someCol");
results.count();
异常发生在方法count()期间。
我还观察到，如果查询是简单的select col from table1，则运行良好，但上面的联接查询会导致错误。
我正
浏览 2提问于2017-03-27得票数 0
回答已采纳
我一直在尝试将1个DataFrame附加到Scala中的另一个DF上。本例中的追加操作只是将一个相同大小的新列添加到现有列中-不涉及键匹配。两个DataFrames的形状相同(仅5行1列)。
scala> val coefficients = lrModel.coefficients.toArray.toSeq.toDF("coefficients")
coefficients: org.apache.spark.sql.DataFrame = [coefficients: double]
scala> coefficients.show()
+---------
浏览 8提问于2017-08-05得票数 0
回答已采纳
我试图在Retrofit上创建一个包装器来抽象我的服务实现。到目前为止，我已经让编译器成功地编译了：
package com.example.spark.testapp.services;
import com.example.spark.testapp.services.apis.Get;
import com.example.spark.testapp.services.apis.Post;
import com.example.spark.testapp.services.utils.*;
import com.example.spark.testapp.services.utils.
浏览 5提问于2016-04-06得票数 15
1回答
众所周知，我们可以在T中初始化模板参数C++的对象。
template <typename T>
struct Cont{
T t;
}
我希望在scala中也这样做，所以我编写了以下代码。还有一些扩展Evaluator的类，EvaluateComponent类似于包装器。我可以编写像new EvaluateComponent[BinaryClassificationEvaluator].evaluate(df)这样的代码
abstract class Evaluator extends Serializable{
var predCol = "predicti
浏览 3提问于2020-04-15得票数 0
回答已采纳
这里是星火2.4.4和德尔塔湖0.5.0。
我正在尝试使用delta数据源创建一个表，而且似乎遗漏了一些东西。尽管CREATE TABLE USING delta命令运行良好，但既没有创建表目录，也没有insertInto工作。
下面的CREATE TABLE USING delta运行良好，但insertInto失败了。
scala> sql("""
create table t5
USING delta
LOCATION '/tmp/delta'
""").show
scala> spark.catalog.li
浏览 4提问于2019-12-31得票数 11
回答已采纳
这是的后续问题。
我希望能够传递Flink的DataSet，并对其执行一些操作，但是数据集的参数是通用的。
我现在遇到的问题是：
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.api.scala._
import scala.reflect.ClassTag
object TestFlink {
def main(args: Array[String]) {
val env = ExecutionEnvironment.getExecutionEnvironment
浏览 2提问于2015-04-09得票数 3
回答已采纳
我想知道在数组中.toList和.toList有什么不同。我在spark-shell中做了这个测试，结果没有区别，但我不知道使用哪个更好。有什么意见吗？
scala> val l = Array(1, 2, 3).toList
l: List[Int] = List(1, 2, 3)
scala> val l = Array(1, 2, 3).to[List]
l: List[Int] = List(1, 2, 3)
浏览 14提问于2019-08-15得票数 3
回答已采纳
4回答
我正在编写一个API，希望能够同时从Scala和Java中使用。我在Scala类中定义了一个多态方法，而且我很难从Java调用它。
根据scala-lang.org上的互操作性，使用使用高级语言特性的Scala类“可能很棘手”，但这并不意味着这是不可能的。由于Scala的高级特性转换为Java的方式没有文档化，它建议拆解类文件来实验它的工作方式。
示例代码
class Polymorphic {
def polyMethod[T](implicit om: Manifest[T]) = {
println(om.erasure.getName)
om.erasure.new
浏览 0提问于2011-11-17得票数 2
回答已采纳
类标记看起来确实很容易使用，但是，从数学上讲，至少在某些情况下(如果不是全部的话)总是可靠地自动插入ClassTag样板难道不可能吗？考虑一下这个例子：
def foo[T: ClassTag] = {
val cls = classTag[T].runtimeClass.asInstanceOf[Class[T]]
val myList: Seq[T] = parseList(rawJson, cls) // let's assume parseList a library method that needs to work with a Class instance
浏览 1提问于2014-02-10得票数 6
回答已采纳
1回答
星火广播大列表无法序列化
、、、、
我有一个很大的字符串列表(140866个元素)，它需要一些时间来计算。计算完后，我想在UDF或DataFrame地图中使用这个列表。我遵循一些教程，并找到了这个例子。
val states = List("NY","New York","CA","California","FL","Florida")
val countries = Map(("USA","United States of America"),("IN","India&
浏览 0提问于2020-11-20得票数 1
回答已采纳
1回答
我有一个带有泛型param的Java接口。
public interface JavaInterface<T> {
void sayHi(T[] keys);
}
我可以实现Java类，比如
public class JavaInterfaceImpl implements JavaInterface<String> {
@Override
public void sayHi(String[] array) {
}
}
但是，当创建Scala类来实现接口时，它失败了
class ScalaInterfaceImpl e
浏览 6提问于2022-08-23得票数 1
1回答
很难获得星火对象的类型参数边界
、、、、
我是Scala的初学者。
我正在尝试创建一个对象，该对象接受ProbabilisticClassifier作为输入，并生成一个CrossValidator模型作为输出：
import org.apache.spark.ml.classification.{ProbabilisticClassifier, ProbabilisticClassificationModel}
import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator
import org.apache.spark.ml.param.ParamMap
imp
浏览 0提问于2018-03-31得票数 1
回答已采纳
我正在尝试在多个列上应用字符串索引器。以下是我的代码
val stringIndexers = Categorical_Model.map { colName =>new StringIndexer().setInputCol(colName).setOutputCol(colName + "_indexed")}
var dfStringIndexed = stringIndexers(0).fit(df3).transform(df3) // 'fit's a model then 'transform's data
for(x<
浏览 77提问于2019-07-22得票数 7
我运行了spark应用程序，连接了两个数据集并形成了一个数据集，并使用编码器将Dataset<Row>转换为Dataset<T>格式。
编码器如下所示：
Encoder<RuleParamsBean> encoder = Encoders.bean(RuleParamsBean.class);
Dataset<RuleParamsBean> ds = new Dataset<RuleParamsBean>(sparkSession, finalJoined.logicalPlan(), encoder);
Dataset<R
浏览 77提问于2018-07-30得票数 1
scala> def foo[U](t: Any) = t.asInstanceOf[U]
foo: [U](t: Any)U
scala> val s: String = foo("hi")
scala> val n = foo("hi")
java.lang.ClassCastException: java.lang.String cannot be cast to scala.runtime.Nothing$
at .<init>(<console>:6)
at .<clinit>(
浏览 1提问于2010-07-01得票数 6
回答已采纳
1回答
我有一个看起来正确的数据集。我可以看到所有行的顺序正确，如下所示： df1.orderBy($"count".desc)
df1.show() 但是当我尝试像这样添加一个限制时： df1.orderBy($"count".desc).limit(5)
df1.show() 我还是得到了所有的行。我想我没看错文件..。[https://spark.apache.org/docs/latest/api/scala/org/apache/spark/sql/Dataset.html#limit(n:Int):org.apache.spark.sql.Dataset[
浏览 10提问于2020-10-09得票数 0
回答已采纳
我正在做DataFrames和DataSets和火花在木星的演习，我遇到了一个问题。当对那些价格超过12万美元的房子执行过滤器时，我发现一个错误，只有当我用Spylon在木星上运行时才会发现这个错误(0.4.1版)，而且我不允许应用过滤器函数接收房子作为参数，而如果我从火花壳终端运行，它会工作，我不明白为什么。附呈的图像和代码：
码
case class House (id: Int, city: String, price: Int)
val houseDF = Seq(House(1,"Paris", 120000),
House(2,"Paris
浏览 0提问于2020-11-17得票数 2
1回答
给定这样一个简单的Scala案例类：
package com.foo.storage.schema
case class Person(name: String, age: Int)
可以从case类创建Spark模式，如下所示：
import org.apache.spark.sql._
import com.foo.storage.schema.Person
val schema = Encoders.product[Person].schema
我想知道是否可以从Python/PySpark中的case类访问模式。我希望能做些类似Python的事情
jvm = sc._jvm
浏览 2提问于2021-01-05得票数 2
回答已采纳
2回答
假设我们有一个嵌套的泛型类：
public class A<T> {
public class B<U> { }
}
在这里，typeof(A<int>.B<>)本质上是一个具有两个参数的泛型类，其中只绑定了第一个参数。
如果我有一个带有两个参数的类
public class AB<T, U> { }
有没有办法提到“使用T=int和U保持开放的AB”？如果不是，这是C#限制还是CLR限制？
浏览 0提问于2011-04-09得票数 18
回答已采纳
我试图在Java的Jackson库的帮助下，从本地到Spark摄取一个JSON文件。模式是在POJO(普通旧Java对象)中定义的。但是，我无法将数据加载到数据集或JavaRDD中。
我试过以下代码，它打印模式，但打印数据显示异常。
public static void main(String[] args) {
//SparkConf conf = new SparkConf().setMaster("spark://master:7077");
SparkConf conf = new SparkConf().setMaster("loca
浏览 0提问于2019-07-20得票数 0
回答已采纳
import org.apache.spark.api.java.JavaRDD
import org.apache.spark.rdd.RDD
import scala.reflect.ClassTag
class TestRDD[T: ClassTag](rdd: RDD[T]) extends JavaRDD(rdd)
从控制台接受此语句。但是，在编译时会引发以下错误：
No ClassTag available for T
[error] class TestRDD[T: ClassTag](rdd: RDD[T]) extends JavaRDD(rdd)
[error]
浏览 3提问于2016-01-21得票数 8
回答已采纳
根据这个，我正在应用udf来过滤CountVectorizer之后的空向量。
val tokenizer = new RegexTokenizer().setPattern("\\|").setInputCol("dataString").setOutputCol("dataStringWords")
val vectorizer = new CountVectorizer().setInputCol("dataStringWords").setOutputCol("features")
val pipeline
浏览 3提问于2018-02-12得票数 1
2回答
我想要创建一个Scala特性，它应该用case类T来实现。这个特性只是加载数据并将其转换为T类型的星火数据集。我得到了一个错误，即不能存储编码器，我认为这是因为Scala不知道T应该是case类。我怎么能告诉编译器呢？我在某个地方见过我应该提到的产品，但是没有这样的类定义。请随意建议其他方法来做这件事！
我有以下代码，但它没有使用错误进行编译: 42: error:无法找到存储在Dataset中的类型的编码器。导入sqlContext.implicits._ INFO .asT支持原始类型(Int、String等)和产品类型(case类)
我用的是星火1.6.1
代码：
import org.
浏览 4提问于2016-11-10得票数 4
回答已采纳
我有两个进程，每个进程做1)连接oracle数据库，读取一个特定的表，2)形成数据帧并处理它。3)将df保存到cassandra。
如果我并行运行这两个进程，则两个进程都尝试从oracle读取数据，而第二个进程读取数据时，我得到以下错误
ERROR ValsProcessor2: org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree:
Exchange SinglePartition
+- *(1) HashAggregate(keys=[], functions=[partial_coun
浏览 0提问于2018-10-26得票数 12
回答已采纳
1回答
在Scala2.10.2中，我创建了一个继承自scala.sys.process.ProcessLogger的类MultiLogger，提供日志复用。
我对这个类的想法看起来有点像这样：
class MultiLogger[T <: ProcessLogger]
(
private val loggers : Set[T]
)
extends Set[T]
with scala.collection.SetLike[T,Set[T]]
with scala.sys.ProcessLogger
{
...
}
它也可能看起来像一组ProcessLogger对
浏览 2提问于2014-05-21得票数 0
我偶然发现了一些带有我不熟悉的泛型符号的示例代码：
vertx.eventBus().<JsonObject>consumer("sensor.updates", message -> {
JsonObject json = message.body();
...
});
注意调用consumer()之前的"“。
我理解它的作用，consumer()接受泛型类型T，我们告诉编译器在第二个参数中要有一个JsonObject。从文档中：
<T> MessageConsumer<T> consumer(String addre
浏览 0提问于2021-05-05得票数 3
3回答
我看过一些类似这样的代码，无法理解它的意义：
public class ClassA{
public <T> void getContactName(ContactList<T> contactList){
//do something
}
}
基本上我不理解这一点。该类编译时没有任何错误。我认为ClassA也应该成为带有参数'T‘的泛型。
谢谢
浏览 0提问于2014-07-31得票数 2
我对Scala很陌生，我不明白这一点：
scala> case class foo[T](value:T){println(value.getClass.toString)}
defined class foo
scala> val bar=foo(1)
class java.lang.Integer
bar: foo[Int] = foo(1)
scala> println(bar.value.getClass.toString)
int
在构造时，它返回'class java.lang.Integer'，当在case类之外查询时，它返回'int
浏览 3提问于2016-01-01得票数 2
回答已采纳
我正在尝试使用spark和esri创建一个带有地理数据集的数据集。如果Foo只有Point字段，它就能工作，但是如果我在Point之外添加了其他字段，就会得到ArrayIndexOutOfBoundsException。
import com.esri.core.geometry.Point
import org.apache.spark.sql.{Encoder, Encoders, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}
object Main {
case class Foo(position:
浏览 4提问于2016-07-21得票数 1
我有两个Scala代码-- MyMain.scala和MyFunction.scala，分别构建和构建的MyFunction jar将在MyMain中充当UDF。
MyFunction.scala基本上包含一个带有公共方法public String myFunc(String val0, String val1)的Java类。该项目是在SBT中构建的，build_jar编译输出作为工件存储(只存储所需的类，即MyFunction.class，而不是依赖项)。
MyMain.scala将上述工件jar导入lib文件夹下，并使用unmanagedBase := baseDirectory.valu
浏览 1提问于2019-10-27得票数 2
回答已采纳
我尝试将PCA应用于我的数据，然后将RandomForest应用于转换后的数据。然而，PCA.transform(数据)给了我一个DataFrame，但我需要一个mllib LabeledPoints来供给我的RandomForest。我该怎么做呢？我的代码：
import org.apache.spark.mllib.util.MLUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.mllib.tree.RandomForest
import org.a
浏览 0提问于2016-03-13得票数 14
回答已采纳
我有一个包含字符串数组的DataFrame列。我尝试过创建一个udf并使用numpy进行置换(unit是列名)：
def permute(row):
return np.random.permutation(row)
udfPermute = udf(permute, ArrayType(StringType()))
print(units.withColumn("shuffled", udfPermute("unit")).head(5))
Py4JJavaError: An error occurred while calling o4246.
浏览 1提问于2017-04-03得票数 0
我正在运行Spark快速入门应用程序： /* SimpleApp.java */
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
public class SimpleApp {
public static void main(String[] args) {
String logFile = "/data/software/spark-2.4.4-bin-without-hadoop/README.md"; // Should be some file
浏览 136提问于2020-01-13得票数 6
1回答
我正在努力使我的graphQL突变成为通用的。但在运行时，我会得到一个错误。
错误是:对象引用未设置为对象的实例。
public class ApplicationMutation<T> : ObjectGraphType where T: BaseModel
{
public ApplicationMutation()
{
this.Name = "RootMutation";
var nameofT = typeof(T).GetElementType().Name;
this.Field<
浏览 7提问于2016-11-24得票数 0
回答已采纳
以下Haskell代码：
main = putStrLn $ "bla " ++ (toStr (A 1) (A 2))
--main2 = putStrLn $ "bla " ++ (toStr (A 1) (A "String")) -- does not compile, as it should
main3 = putStrLn $ "bla " ++ (toStr (A "String") (A "String"))
data A a = A a deriving Show -- (1)
浏览 1提问于2017-04-03得票数 2
回答已采纳
1回答
我想从class方法中添加列，但遇到了一些错误。 首先，这是我的数据框架 from pyspark.sql.functions import udf
import pandas as pd
df = spark.createDataFrame(pd.DataFrame([[1,1,1],[2,2,2]],columns=['a','b','c']))
+---+---+---+
| a| b| c|
+---+---+---+
| 1| 1| 1|
| 2| 2| 2|
+---+---+---+ 然后，我创建类方法 c
浏览 22提问于2019-04-09得票数 0
2回答
我有两个数据集，可以单独查询和显示()。一张有17张唱片，另一张有3张。
Dataset<Row> attReader = spark
.read()
.format("org.apache.spark.sql.cassandra")
.option("table", "table_1")
.load();
Dataset<Row> surReader = spark
.read()
.format("org.apache.spark.sql.cassandra&#
浏览 0提问于2018-10-08得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
