Java Spark Dataset MapFunction -如果没有任何对类的引用，则任务不可序列化
Java Spark Dataset MapFunction是Spark框架中的一个接口，用于对Dataset进行映射操作。MapFunction接口定义了一个方法，即call()方法，用于将输入的数据集中的每个元素进行转换并返回新的元素。
在Spark中，Dataset是一种强类型的分布式数据集，可以看作是一组被分区的数据集合，每个分区存储在集群中的不同节点上。通过使用MapFunction，可以对Dataset中的每个元素进行处理，实现数据的转换、过滤、计算等操作。
对于Java Spark Dataset MapFunction，如果没有任何对类的引用，则任务不可序列化。这是因为在Spark中，任务需要被序列化并在集群中传输，而如果任务中存在对类的引用，那么这个类必须是可序列化的，否则会导致任务无法正确传输和执行。
为了解决这个问题，可以将MapFunction的实现类定义为静态内部类或顶级类，或者使用匿名内部类的方式实现MapFunction接口。这样可以避免对外部类的引用，使任务能够正确序列化。
在Spark中，MapFunction常用于对Dataset进行数据转换和处理，例如对每个元素进行映射、提取特定字段、计算新的字段等操作。通过使用MapFunction，可以灵活地对数据进行处理，满足不同的业务需求。
推荐的腾讯云相关产品和产品介绍链接地址：
- 腾讯云Spark：https://cloud.tencent.com/product/spark
- 腾讯云大数据计算服务：https://cloud.tencent.com/product/dc
- 腾讯云云服务器CVM：https://cloud.tencent.com/product/cvm
- 腾讯云云数据库CDB：https://cloud.tencent.com/product/cdb
- 腾讯云人工智能AI：https://cloud.tencent.com/product/ai
- 腾讯云物联网IoT Hub：https://cloud.tencent.com/product/iothub
- 腾讯云移动开发MPS：https://cloud.tencent.com/product/mps
- 腾讯云对象存储COS：https://cloud.tencent.com/product/cos
- 腾讯云区块链BCS：https://cloud.tencent.com/product/bcs
- 腾讯云元宇宙：https://cloud.tencent.com/product/mu
相关·内容
我有一个下面的类，它将csv数据读取到Spark的Dataset中。如果我只是简单地读取并返回data，一切都会正常工作。
但是，如果在从函数返回之前对data应用MapFunction，则会得到
Exception in thread "main" org.apache.spark.SparkException: Task not serializable
Caused by: java.io.NotSerializableException: com.Workflow。
我知道Spark的工作原理和它序列化对象以进行分布式处理的需求，然而，我没有在映射逻辑中使用任何对Wor
浏览 83提问于2021-02-17得票数 0
回答已采纳
关于kryo序列化，我错过了什么？
Class1和Class3不是java可序列化的类(没有默认的构造函数，也没有getter和setter)
当我尝试“使用”一个实例(它是在星火上下文中创建的)时，无论我是否将Classe3注册为Kryo类，我都会遇到一个序列化问题。
工作良好：
Dataset<Class1> ds = spark.createDataset(classes, Encoders.kryo(Class1.class));
Dataset<String> df = df.map((MapFunction<Class1, String>) c
浏览 0提问于2019-02-22得票数 1
回答已采纳
1回答
我们需要像jaccard这样的距离矩阵来计算星星之火中的海量数据集。面对几个问题。请帮助我们给出指示。
第1期
import info.debatty.java.stringsimilarity.Jaccard;
//sample Data set creation
List<Row> data = Arrays.asList(
RowFactory.create("Hi I heard about Spark", "Hi I Know about Spark"),
浏览 2提问于2017-03-09得票数 0
1回答
闪烁序列化错误
、、
我正在尝试在我的Apache Flink Gelly图上运行标签传播协议。
下面是我的代码：
Graph<String, Long, String> ugraph = Graph.fromDataSet(vertex, edgeSet, env).getUndirected();
DataSet<Tuple2<String, Long>> idsWithInitialLabels = DataSetUtils
.zipWithUniqueId(graph.getVertexIds())
浏览 2提问于2016-07-27得票数 1
def URLEnc(input: String): String = {
URLEncoder.encode(input, "UTF-8")
}
val URLEncUDF: UserDefinedFunction = udf(URLEnc(_: String))
val file = spark.read.format("xml")
.option("rootTag", "channel").option("rowTag", "item")
.load("
浏览 0提问于2020-12-16得票数 0
我已经使用下面指定的withColumn尝试了这个解决方案： How to cast all columns of Spark dataset to string using Java 但是，解决方案会影响大量列(1k-6k)的性能。它需要超过6个小时，然后被中止。 或者，我尝试使用map进行强制转换，如下所示，但我在这里得到错误： MapFunction<Column, Column> mapFunction = (c) -> {
return c.cast("string");
};
dataset = dataset.map(ma
浏览 29提问于2020-10-22得票数 0
回答已采纳
1回答
我正在做一个Flink项目，遇到了一个问题，我在Stackoverflow的回答下设法解决了这个问题。然而，我不清楚为什么提出的解决方案实际上是有效的，我发现有关这个主题的信息很少。考虑以下代码：
object DeCP {
def main(args: Array[String]): Unit = {
val params: ParameterTool = ParameterTool.fromArgs(args)
// Get the execution environment and read the data
val env: ExecutionEnviro
浏览 1提问于2018-09-18得票数 3
回答已采纳
在flink中，要将变量序列化为节点，可以将参数作为运算符中的参数传递 env = ...map(new MapFunction(param)) 也可以被广播 dataset.map(...).withBroadcastSet() 实际上，param和dataset都是程序中的变量，没有文字上的区别。 那么，这些方法到底有什么不同呢？
浏览 31提问于2020-08-14得票数 0
回答已采纳
这是我在SO上的第一篇文章，如果使用了不正确的格式，我深表歉意。
我正在与Apache Spark合作创建一个新的源代码(通过DefaultSource)，BaseRelations等……并且遇到了序列化的问题，我想更好地了解它。下面考虑一个扩展BaseRelation并实现扫描构建器的类。
class RootTableScan(path: String, treeName: String)(@transient val sqlContext: SQLContext) extends BaseRelation with PrunedFilteredScan{
privat
浏览 13提问于2017-01-14得票数 3
回答已采纳
我正在尝试执行下面的火花流示例代码。
我正在一个AWS EMR集群上运行它，它的版本为SparkVersion2.0.2。下面的依赖项被添加到星火提交中。
spark-sql-kafka-0-10_2.11-2.0.2.jar
spark-streaming-kafka-0-8-assembly_2.11-2.0.2.jar
kafka-clients-0.10.2.0.jar
下面是错误日志：
Batch: 0
-------------------------------------------
17/06/06 19:44:01 WARN TaskSetManager
浏览 2提问于2017-06-06得票数 0
回答已采纳
1回答
我正在尝试从一个蜂巢表中获取记录，但是当尝试进行选择时，它会说该表不存在。错误说是Table or view not found: clientes lines 1 pos 14
import java.io.File;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apach
浏览 0提问于2018-05-09得票数 2
回答已采纳
我在Spark ( 2.2.0 )中看到了一种我不明白的行为，但当试图提取Lambda函数时，我猜想它与lambda和匿名类有关：
这样做是可行的：
public class EventsFilter
{
public Dataset< String > filter( Dataset< String > events )
{
return events.filter( ( FilterFunction< String > ) x -> x.length() > 3 );
}
}
然而，这并不是：
publ
浏览 10提问于2017-09-17得票数 1
回答已采纳
下面的java程序是用apache spark编写的。
该程序尝试从相应的文件中读取肯定和否定词的列表，将其与主文件进行比较，并相应地过滤结果。
import java.io.Serializable;
import java.io.FileNotFoundException;
import java.io.File;
import java.util.*;
import java.util.Iterator;
import java.util.List;
import java.util.List;
import org.apache.spark.api.java.*;
import org
浏览 0提问于2014-09-18得票数 20
回答已采纳
我正在将测试从JUnit4迁移到JUnit5。代码在JUnit4中运行良好，但在JUnit 5中，它会引发org.apache.spark.SparkException: Task not serializable异常。我试过多种方法，但现在起作用了。在testAlternativeVegetableIdWithDifferentReadCount行中，assertTrue测试失败。当我试图显示Dataset值时，它也会抛出一个错误。我做错什么了？
@ExtendWith(MockitoExtension.class)
public class ExportLogicTest implemen
浏览 5提问于2022-07-11得票数 0
我是新手，并且尝试运行示例JavaSparkPi.java，它运行得很好，但是由于我必须在另一个java中使用它，所以我将所有东西从main复制到类中的一个方法，然后尝试在main中调用该方法，它可以帮助
org.apache.spark.SparkException:作业中止:不可序列化的任务: java.io.NotSerializableException
代码如下所示：
public class JavaSparkPi {
public void cal(){
JavaSparkContext jsc = new JavaSparkContext("local
浏览 1提问于2014-06-04得票数 12
回答已采纳
当尝试使用Spark为应用程序创建测试时，我面临以下错误：
java.io.InvalidClassException: java.lang.Void; local class name incompatible with stream class name "void"
at java.io.ObjectStreamClass.initNonProxy(ObjectStreamClass.java:620)
at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1843)
浏览 0提问于2019-06-13得票数 0
回答已采纳
1回答
如何将映射转换为列
、、、
我用的是笔记本。所以火花基本上是在互动模式下运行的。这里我不能使用闭包变量，因为齐柏林飞艇抛出了org.apache.spark.SparkException: Task not serializable，因为它试图序列化整个段落(更大的闭包)。
因此，如果没有闭包方法，我只能将map作为列传递给UDF。
我收集了一张从已销毁的RDD中收集的地图：
final val idxMap = idxMapRdd.collectAsMap
它正被用于星火变换中：
def labelStr(predictions: WrappedArray[Double], idxMap: Map[Double, St
浏览 3提问于2016-11-14得票数 0
当我尝试训练线性回归模型时，我得到了下面的异常(然而，当我使用单独的JVM训练模型时，同样的事情也可以正确执行)：
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 13.0 failed 4 times, most recent failure: Lost task 0.3 in stage 13.0 (TID 28, impetus-dsrv07.impetus.co.in, executor 2): java.lang.ClassCastException: cannot
浏览 103提问于2017-12-20得票数 0
我有一个小的Scala代码，它可以在Spark-Shell上正常工作，但不能在带有Scala插件的Eclipse中工作。我可以使用插件访问hdfs试着写另一个文件，它工作了..
FirstSpark.scala
package bigdata.spark
import org.apache.spark.SparkConf
import java. io. _
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
object FirstSpark {
def main(args: Arra
浏览 8提问于2016-07-20得票数 0
回答已采纳
我想不断地详细说明dataset流的行(最初由Kafka启动)：基于一个条件，我想更新Radis哈希。这是我的代码片段(lastContacts是前一个命令的结果，该命令是这种类型的流：org.apache.spark.sql.DataFrame = [serialNumber: string, lastModified: long] )。这将扩展到org.apache.spark.sql.Dataset[org.apache.spark.sql.Row])：
class MyStreamProcessor extends ForeachWriter[Row] {
override def
浏览 0提问于2019-03-26得票数 0
回答已采纳
3回答
序列化和自定义Spark类
、、、、
我正在Scala中编写一个定制的Spark实现，并且正在使用shell调试我的实现。我现在的目标是：
customRDD.count
毫无例外地取得成功。现在我得到的是：
15/03/06 23:02:32 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks
15/03/06 23:02:32 ERROR TaskSetManager: Failed to serialize task 0, not attempting to retry it.
java.lang.reflect.InvocationTargetException
浏览 0提问于2015-03-06得票数 1
1回答
我有以下Java代码，可以从HDFS读取JSON文件，并使用Spark将其输出为配置单元视图。
package org.apache.spark.examples.sql.hive;
import java.io.File;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apac
浏览 14提问于2018-01-23得票数 0
回答已采纳
我试图使用Spark连接器将数据保存到大型查询中。假设我有一个像下面这样的Java pojo
@Getter
@Setter
@AllArgsConstructor
@ToString
@Builder
public class TagList {
private String s1;
private List<String> s2;
}
现在，当我试图将这个Pojo保存到大型查询中时，它会将我抛到错误下面
Caused by: com.google.cloud.spark.bigquery.repackaged.com.google.cloud.bigquery.
浏览 0提问于2020-11-26得票数 0
回答已采纳
1回答
我试图使用Foreach分区来迭代一个数据文件，以便将一个值插入到数据库中。我使用了前端分区，对行进行分组，并使用foreach迭代每一行。请在下面找到我的密码，
val endDF=spark.read.parquet(path).select("pc").filter(col("pc").isNotNull);
endDF.foreachpartition((partition: Iterator[Row]) =>
class.forname(driver)
val con=DriverManager.connection(jdbcu
浏览 0提问于2021-12-02得票数 0
DataSet person = spark.read.textfile(path).map(Person::new，Encoders.bean(Person.class))
当我尝试上面的工作时，它将在spark2.4(scala-2.11)中工作，但在scamk3.1.1(Scala-2.12)中，它显示为DataSet类型的矛盾性。在我使用地图的地方，过滤器，地图分区，平面图也会有矛盾/
但是，通过进行类型转换，spark.read.textfile(path).map((MapFunction<String，Person>(Person：：new，Encoders.bean
浏览 8提问于2021-12-13得票数 0
我是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
回答已采纳
..。
val cols: Seq[String] = Seq("item", "SR", "RP")
val vecToSeq = udf((v:org.apache.spark.ml.linalg.Vector) => v.toArray)
val exprs = cols.zipWithIndex.map{ case(c,i) => $"_tmp".getItem(i).alias(c)}
val DoubleDF = result5.select(vecToSeq($"vectorCol").
浏览 0提问于2017-02-28得票数 0
正在尝试对dataframe使用show()方法。它正在给Task not serializable异常。
我试图扩展Serializable对象，但错误仍然存在。
object App extends Serializable{
def main(args: Array[String]): Unit = {
Logger.getLogger("org.apache").setLevel(Level.WARN);
val spark = SparkSession.builder()
.appName("LearningSpark
浏览 0提问于2019-09-01得票数 2
当我在我的spark(用java编写)应用程序中使用UDF函数时，我得到了这个错误。
org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:403)：任务在org.apache.spark.SparkException处不可序列化...由:类: jp.co.nec.necdas.commons.customize.service.dataset.ALMTriggerProcessLogic序列化堆栈：-对象不可序列化引起(类: jp.co.nec.necdas.commons.customiz
浏览 0提问于2019-10-17得票数 0
我一直在尝试在py函数之外的pyspark中实现udf，如下所示： 它采用了我之前训练过的bin模型。 def sentiment(frase):
classifier = load_model("sentiment_fasttext.bin")
sentiment = classifier.predict(frase)
sentiment = ''.join(map(str, sentiment))
return sentiment
sentiment = df.withColumn('sentiment',
浏览 15提问于2020-09-23得票数 0
1回答
我试图在intellij idea中实现case类，我得到了以下错误。你能帮我解决这个问题吗？
代码：
val conf = new SparkConf().setMaster("local").setAppName("case class")
val sc = new SparkContext(conf)
case class Employee (empno: String, ename: String, job: String, mgr: String,
hiredate: String, sal: String, comm: St
浏览 2提问于2019-12-17得票数 1
我的问题与Scala非常相似，只是我使用的是this。 对于下面的代码： roleList = res.select($"results", explode($"results").as("results_flat1"))
.select("results_flat1.*")
.select(explode($"rows").as("rows"))
浏览 38提问于2019-12-11得票数 0
回答已采纳
我正在编写一个应用程序，将消息从Kafka读取到Spark使用结构化流。
传入消息是字符串格式的与交易相关的修复消息。它们被转换为Java POJO。
示例代码如下：
SparkSession spark = createSparkSession();
// Subscribe to 1 topic
Dataset<Row> df = spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", brokers)
浏览 2提问于2017-08-22得票数 1
我开始玩星火2.0.1了。新的Dataset API非常干净，但是我在非常简单的操作中遇到了问题。
也许我错过了什么希望有人能帮上忙。
这些指示
SparkConf conf = new SparkConf().setAppName("myapp").setMaster("local[*]");
SparkSession spark = SparkSession
.builder()
.config(conf)
.getOrCreate();
Dataset<Info> infos = spark.r
浏览 11提问于2016-11-11得票数 4
我对Java 8和Spark2.1.1有问题
我有一个(有效的)正则表达式保存在一个名为"pattern“的变量中。当我尝试使用这个变量过滤从文本文件中加载的内容时，会抛出一个SparkException :任务不可串行化。有谁可以帮我？以下是代码：
JavaRDD<String> lines = sc.textFile(path);
JavaRDD<String> filtered = lines.filter(new Function<String, Boolean>() {
@Override
publi
浏览 0提问于2018-08-18得票数 2
回答已采纳
我有两个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
回答已采纳
3回答
我与卡桑德拉使用火花，我有一个客户的JavaRDD<String>。对于每个客户，我想从cassandra中选择这样的交互方式：
avaPairRDD<String, List<InteractionByMonthAndCustomer>> a = client.mapToPair(new PairFunction<String, String, List<InteractionByMonthAndCustomer>>() {
@Override
public Tuple2<String, Lis
浏览 4提问于2014-12-30得票数 7
回答已采纳
1回答
我在Spark2.2.0中有一个不可序列化的类异常。以下是我在Scala中要做的工作：
从HDFS读取一组JPEG图像。
要构建java.awt.image.BufferedImageS数组，请执行以下操作。
提取java.awt.image.BufferedImage缓冲区并将其存储在每个图像的2D数组中，方法是构建一个包含图像缓冲区信息Array[Array[Int]]的二维数组。
使用Array[Array[Int]]方法将org.apache.spark.rdd.RDD[Array[Array[Int]]]转换为org.apache.spark.rdd.RDD[Arra
浏览 7提问于2018-01-16得票数 1
回答已采纳
1回答
如何在Worker上使用瞬态变量
、、、、
我想知道瞬态变量在工人身上是如何可用的。例如：-通过串行化MapFunction对象，从驱动程序向执行器发送映射任务命令。executor反序列化命令，并在分区上执行它。现在，如果我在mapFunction中使用了一个瞬态变量，它如何在工作进程上可用，因为它没有被序列化并发送给工作进程。
同样在下面的链接的示例中
示例：
Class Test{
transient static SparkSession sparkSession;
public static void main(String[] args){
sparkSession = //Initialize SparkS
浏览 5提问于2016-08-29得票数 2
1回答
我想在flink流处理作业中将对象序列化为Json。在flink作业中，所有对象都需要可序列化。
我用杰克逊来序列化我的对象。
final ObjectMapper mapper = new ObjectMapper();
mapper.registerModule(new JodaModule());
mapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
当我使用它时，我会得到以下例外：
java.io.NotSerializableException: com.fasterxml.jackson.datatype.j
浏览 8提问于2016-12-08得票数 0
回答已采纳
该错误可以在火花壳中再现。基本上，我定义了一个具有生成RDD的方法的类，然后在RDD上执行一个映射操作，这将生成序列化错误。
如果我没有方法，只有执行方法步骤的语句，那么一切都正常。
代码在这里，可以在shell中运行，我定义了一个类，然后实例化这个类。
首先是进口
import java.nio.file.{Files}
import java.io._
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import scala.io.Source
import scala.collection.muta
浏览 1提问于2015-08-18得票数 4
我有一个这样的数据：
val temp = sc.parallelize(Seq(Array(43,53,266),Array(69,160,166),Array(266)))
.toDF("value")
我想选择与以下数组相交的行：
val goodValue = Array(231, 266)
val broadcastGood = sc.broadcast(goodValue)
val containGood = udf((array:scala.collection.mutable.WrappedArray[Int]) =>
broadcastGood.val
浏览 1提问于2017-06-25得票数 1
1回答
我有一个使用Databricks笔记本从数据集创建的RDD。
当我试图从它获得具体的值时，它只是在序列化错误消息中失败。
这里是我获取数据的地方(PageCount是一个Case类)：
val pcDf = spark.sql("SELECT * FROM pagecounts20160801")
val pcDs = pcDf.as[PageCount]
val pcRdd = pcDs.rdd
当我这么做时：
pcRdd.take(10)
我得到以下例外：
org.apache.spark.SparkException: Job aborted due to stage f
浏览 1提问于2016-10-29得票数 4
我对Java 8和它的lambda有多年的经验。但我遇到了一个疯狂的问题，当我开发了一个世界大小的星火计划。
这里有一个Java类，其中数据注释来自Lombok：
@Data
public class Person implements Serializable {
private String name;
private Long age;
}
然后我构建了一个包含Persion类对象的java列表：
Person p1 = new Person("sb", 1L);
Person p2 = new Person("sth
浏览 5提问于2021-11-24得票数 2
回答已采纳
在我的Spark代码中，我尝试从csv文件创建一个IndexedRowMatrix。但是，我得到以下错误：
Exception in thread "main" org.apache.spark.SparkException: Task not serializable
...
Caused by: java.io.NotSerializableException: org.apache.spark.api.java.JavaSparkContext
下面是我的代码：
sc = new JavaSparkContext("local", "App"
浏览 3提问于2015-06-14得票数 5
2回答
我有一堆带有DTD头的xml，我试图加载所有这些文件，而忽略了DTD。
val input = sc.wholeTextFiles("""\path\*.nxml""")
val saxfac = SAXParserFactory.newInstance();
saxfac.setValidating(false);
saxfac.setFeature("http://xml.org/sax/features/validation", false);
saxfac.setFeature("http://apache.or
浏览 0提问于2015-01-13得票数 0
回答已采纳
我遇到了Java Spark Dataset的groupByKey方法的问题。以下代码在测试环境(Spark 2.1.0、spark-core_2.11、spark-sql_2.11)中本地运行时引发以下异常：
java.lang.Exception: failed to compile: org.codehaus.commons.compiler.CompileException: File 'generated.java', Line 43, Column 21: No applicable constructor/method found for zero actual
浏览 0提问于2017-05-24得票数 1
我在理解Java中的火花函数实现时遇到了麻烦。给出了在map和reduce中使用函数的三种方法：
拉姆达
通过内联类实现Function和Function2
通过内部类实现Function和Function2
问题是我无法使2.和3.工作。例如，这个代码：
public int countInline(String path) {
String master = "local";
SparkConf conf = new SparkConf().setAppName("charCounterInLine")
浏览 3提问于2014-07-09得票数 7
回答已采纳
我有一个简单的模型类
@Value // lombok - create standard all arg constructor and getters
public class ModelA implements Serializable {
private String word;
private double value;
}
这个简单的测试失败了：
public class SparkSerializationTest {
private SparkSession spark = SparkSession.builder()
.mas
浏览 4提问于2020-01-06得票数 0
回答已采纳
下面我提供了我的代码。我遍历了DataFrame prodRows，对于每个product_PK，我从prodRows中找到了一些匹配的product_PKs子列表。
numRecProducts = 10
var listOfProducts: Map[Long,Array[(Long, Int)]] = Map()
prodRows.foreach{ row : Row =>
val product_PK = row.get(row.fieldIndex("product_PK")).toString.toLong
val gend
浏览 3提问于2017-11-18得票数 5
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
