Spark:2.0.2 java.util.ConcurrentModificationException: KafkaConsumer对于多线程访问是不安全的
Spark是一个开源的大数据处理框架，它提供了高效的数据处理和分析能力。Spark支持多种编程语言，包括Java、Scala、Python和R等。它的核心概念是弹性分布式数据集（Resilient Distributed Dataset，简称RDD），它是一个可并行操作的分布式对象集合。
在Spark中，java.util.ConcurrentModificationException是一个常见的异常，表示在迭代集合的过程中，其他线程对集合进行了修改，导致迭代器抛出异常。对于KafkaConsumer来说，它是Kafka消息队列的消费者，用于从Kafka主题中读取消息。
由于KafkaConsumer是非线程安全的，即不能在多个线程中共享同一个KafkaConsumer实例。如果多个线程同时访问同一个KafkaConsumer实例，就会导致java.util.ConcurrentModificationException异常。
为了解决这个问题，可以采用以下两种方式之一：
- 每个线程使用独立的KafkaConsumer实例：每个线程创建自己的KafkaConsumer实例，并独立消费消息。这样可以避免多个线程之间的竞争和冲突。
- 使用线程安全的KafkaConsumer实现：某些第三方库或框架提供了线程安全的KafkaConsumer实现，可以在多线程环境中使用。例如，Apache Kafka提供了一个名为KafkaConsumerThreadSafe的线程安全实现。
在腾讯云的产品中，可以使用腾讯云的消息队列CMQ（Cloud Message Queue）来替代Kafka，CMQ提供了高可靠、高可用的消息队列服务，适用于大规模分布式系统的消息通信。您可以通过腾讯云CMQ的官方文档了解更多信息：腾讯云CMQ产品介绍
总结：对于Spark中的java.util.ConcurrentModificationException异常，可以通过每个线程使用独立的KafkaConsumer实例或使用线程安全的KafkaConsumer实现来解决。腾讯云提供了CMQ作为替代方案，用于实现可靠的消息队列服务。
相关·内容
我收到以下错误：
java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
当我在代码中包含这段代码时：
CuratorFrameworkFactory.newClient()
我不明白是什么导致了这个错误。有什么帮助吗？
谢谢。
浏览 1提问于2018-12-19得票数 4
我使用kafka 0.10.1.0和spark 2.0.2得到以下错误
private val spark = SparkSession.builder()
.master("local[*]")
.appName(job.name)
.config("spark.cassandra.connection.host","localhost"))
.config("spark.cassandra.connection.port","9042")
.config("spark.streaming.receive
浏览 10提问于2016-12-20得票数 5
我使用下面的代码来读取Kafka主题，并处理数据。
JavaDStream<Row> transformedMessages = messages.flatMap(record -> processData(record))
.transform(new Function<JavaRDD<Row>, JavaRDD<Row>>() {
//JavaRDD<Row> records = ss.emptyDataFrame().toJavaRDD();
浏览 1提问于2017-06-14得票数 11
3回答
kafka的医生给出了一种方法，描述如下：
每线程一个使用者:一个简单的选项是给每个线程自己的使用者>实例。
我的代码：
public class KafkaConsumerRunner implements Runnable {
private final AtomicBoolean closed = new AtomicBoolean(false);
private final CloudKafkaConsumer consumer;
private final String topicName;
public KafkaConsumerR
浏览 3提问于2016-03-29得票数 10
回答已采纳
我使用的是Spark流媒体应用程序。应用程序使用直接流从Kafka topic (具有200个分区)中读取消息。应用程序偶尔会抛出ConcurrentModificationException->
java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:1431)
at org.apache.kafka
浏览 1提问于2017-12-03得票数 2
回答已采纳
1回答
当经纪公司倒闭时，我在卡夫卡客户身上看到了这个例外：
java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:2452)
at org.apache.kafka.clients.consumer.KafkaConsumer.acquireAndEnsureOpen(KafkaConsumer.
浏览 4提问于2021-02-21得票数 0
回答已采纳
当尝试使用Apache 1.6.3启用Kerberos启用Apache (0.9)时出错。动物园管理员版本为3.4.5，我必须连接到两个kafka。一种是启用keberos，另一种则不是，所以我不会在spark的额外java选项中设置java.security.auth.login.config属性。
Kafka Initialization failed: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
at org.apache.kafka.clients.consumer.Kaf
浏览 4提问于2017-08-01得票数 0
当将Kafka消息转换为dataframe时，在将包作为参数传递时会出现错误。
from pyspark.sql import SparkSession, Row
from pyspark.context import SparkContext
from kafka import KafkaConsumer
import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--jars spark-sql-kafka-0-10_2.11-2.0.2.jar,spark-streaming-kafka-0-8-assembly_2.11-2.
浏览 5提问于2020-06-12得票数 1
1回答
当我提交星火申请时，会得到以下错误：
线程“流启动”java.lang.NoSuchMethodError中的异常：
org.apache.kafka.clients.consumer.KafkaConsumer.subscribe(Ljava/util/Collection;)V
浏览了以下网址：
设置以下环境变量：SPARK_KAFKA_VERSION=0.10
添加了以下依赖项：
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-stream
浏览 1提问于2017-08-23得票数 1
我使用的是星星之火-SQL2.4.1、星火-卡桑德拉-连接器_2.11-2.4.1.jar和java8。同时将数据从kafka主题插入到C*/Cassandra表数据。
我正在犯错误：
org.apache.spark.sql.kafka010.KafkaDataConsumer - KafkaConsumer cache hitting max capacity of 64, removing consumer for CacheKey(spark-kafka-source-33321dde-bfad-49f3-bdf7-09f95883b6e9--1249540122-executor)
浏览 0提问于2019-10-18得票数 0
我正在运行下面的代码，在从Kafka读取了几个RDDs之后，我得到了这个Kafka多线程访问错误。不确定是什么导致了这种情况。基本上，代码所做的是从kafka读取CDR，并尝试将其放入全局临时视图中。简而言之，所有的RDDs /数据帧都必须存储在全局视图中。我使用的是Kafka 0.10和spark 2.1.1
val schema_string = "subscriberNumber, originNodeType, originHostName, originOperatorID, originTimeStamp, currentServiceClass, voucherBase
浏览 0提问于2018-05-06得票数 1
2回答
我已经看到了一个类似于的问题
但是，我仍然想知道来自特定分区的流数据是否不可能实现？我使用了卡夫卡消费策略在星火流订阅方法。
ConsumerStrategies.Subscribe
这是我尝试订阅主题和分区的代码片段，
val topics = Array("cdc-classic")
val topic="cdc-classic"
val partition=2;
val offsets=
Map(new TopicPartition(topic, partition) -> 2L)//I am not clear with this lin
浏览 0提问于2018-06-07得票数 0
回答已采纳
我有一个Kafka消费者，它从主题中读取消息，并使用spark将其写入到hive表中。当我在Yarn上运行代码时，它会多次读取相同的消息。我在这个主题中有大约100,000条消息。但是，我的消费者仍然多次阅读相同的内容。当我做distinct时，我会得到实际的计数。 这是我写的代码。我想知道我是否遗漏了任何设置。 val spark = SparkSession.builder()
.appName("Kafka Consumer")
.enableHiveSupport()
.getOrCreate()
import spar
浏览 16提问于2020-07-22得票数 2
2回答
创建props对象后设置使用者的代码
val consumer = new KafkaConsumer[String, String](props)
consumer.subscribe(util.Arrays.asList(topic))
代码导入如下所示
package main.scala
import org.apache.spark.SparkContext
import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.strea
浏览 2提问于2017-08-11得票数 0
1回答
我的代码是：
SparkSession spark = SparkSession.builder().appName("plzzzz").master("local[*]").getOrCreate();
Dataset<Row> dataset = spark.read().format("libsvm").load("/home/ceny/tools/123.txt");
KMeans kmeans = new KMeans().setK(2).setSeed(1L);
KMeansModel model = km
浏览 5提问于2016-12-18得票数 0
回答已采纳
我用火花流来消耗卡夫卡的原型信息。
当主服务器被设置为"local2“时，它工作得很好，但是当我将主url更改为一个真正的星星团的主url时，我会遇到以下异常
Exception in thread "main" org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 20.0 failed 4 times, most recent failure: Lost task 0.3 in stage 20.0 (TID 58, 10.0.5.155): java.l
浏览 6提问于2016-12-07得票数 1
回答已采纳
尝试使用Pyspark2.0.2-hadoop2.7在与Python2.7.x一起使用时出现错误
代码为：
import os
import sys
os.environ['SPARK_HOME']="C:/Apache/spark-2.0.2-bin-hadoop2.7"
sys.path.append("C:/Apache/spark-2.0.2-bin-hadoop2.7/python")
try:
from pyspark import SparkContext
from pyspark import SparkCon
浏览 0提问于2016-11-24得票数 0
3回答
我是星火流的新手，从星火流到卡夫卡，我使用KafkaUtils创建了一个直接流。就像下面
jssc = new JavaStreamingContext(sparkConf, Durations.seconds(KafkaConfig.getInstance().getBatchDuration()));
HashMap<String, String> kafkaParams = new HashMap<>();
kafkaParams.put(KafkaConstants.ZOOKEEPER_CONNECTION_STRING, zookeeperHost + &
浏览 4提问于2017-11-15得票数 0
所以使用：- Spark Structured Streaming (2.1.0) - Kafka 0.10.2.0 - Scala 2.11
我使用的是Kafka的默认API，所以基本上：
val df = spark.readStream
.format("kafka")
.option(...)
设置选项(通过SSL)和所有内容。然后，我直观地应用了一些操作等，并启动了流等(它可以正常运行)。然而，它会不时地抛出一个异常：
17/05/30 11:05:23 WARN TaskSetManager: Lost task 23.0 in stage 77.0 (TI
浏览 4提问于2017-05-30得票数 1
我试图运行电火花流应用与卡夫卡使用纱线。我得到了下面的堆栈跟踪错误-
原因是: org.apache.kafka.common.config.ConfigException:缺少没有默认值的必需配置"partition.assignment.strategy“。在org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:124) at org.apache.kafka.common.config.AbstractConfig.(AbstractConfig.java:48) at org.apache.kafka.cl
浏览 0提问于2019-03-13得票数 3
1回答
我开创了卡夫卡，创造了一个主题和一个制片人。现在我想看看那个制片人发来的消息。我的代码
def main(args: Array[String]){
val sparkConf = new SparkConf()
val spark = new SparkContext(sparkConf)
val streamingContext = new StreamingContext(spark, Seconds(5))
val kafkaStream = KafkaUtils.createStream(streamingContext
, "l
浏览 5提问于2017-11-17得票数 0
回答已采纳
我正在尝试用Java做一个简单的Spark SQL编程。在程序中，我从Cassandra表中获取数据，将RDD转换为Dataset并显示数据。当我运行spark-submit命令时，我得到了错误：java.lang.ClassNotFoundException: org.apache.spark.internal.Logging。
我的程序是：
SparkConf sparkConf = new SparkConf().setAppName("DataFrameTest")
.set("spark.cassandra.connection.host
浏览 0提问于2016-12-06得票数 5
在Spark 1.6.2中，可以通过以下方法访问HiveConf对象。
有人能告诉我在Spark2.0.2中如何从SparkSession对象做同样的事情吗？
浏览 3提问于2016-12-22得票数 2
1回答
我有一个JSON输入文件：
{"a": "abc", "b": "bcd", "d": 3},
{"a": "ezx", "b": "hdg", "c": "ssa"},
...
每个对象都缺少一些字段，而不是放置null值。
在使用Scala的Apache Spark中：
import SparkCommons.sparkSession.implicits._
private val inputJsonPath: S
浏览 4提问于2017-09-23得票数 4
环境: Spark 2.3.0，Scala2.11.12，Kafka (不管最新版本是什么)
我有一个安全的卡夫卡系统，我试图连接我的星火流消费者。下面是我的build.sbt文件：
name := "kafka-streaming"
version := "1.0"
scalaVersion := "2.11.12"
// still want to be able to run in sbt
// https://github.com/sbt/sbt-assembly#-provided-configuration
run in Comp
浏览 3提问于2019-06-10得票数 9
目前，我正在将Cassandrarow RDD转换为dataframe：
val ssc = new StreamingContext(sc, Seconds(15))
val dstream = new ConstantInputDStream(ssc, ssc.cassandraTable("db", "table").select("createdon"))
import sqlContext.implicits._
dstream.foreachRDD{ rdd =>
val dataframeJobs = rdd.m
浏览 5提问于2017-05-30得票数 0
回答已采纳
我是卡夫卡的新手，我通过了，但我什么都不懂。有人能解释一下什么时候使用ConcurrentKafkaListenerContainerFactory类吗？我已经使用了Kafkaconsumer类，但是我发现在我当前的项目中正在使用ConcurrentKafkaListenerContainerFactory。请解释一下它的用途。
浏览 2提问于2019-03-06得票数 11
回答已采纳
The image shows 8081 UI.主程序显示了当我启动shell或pyspark时正在运行的应用程序。但是，当我使用spark-submit运行python脚本时，master并没有显示任何正在运行的应用程序。这是我使用的命令：spark-submit --master spark://localhost:7077 sample_map.py。网络用户界面为:4040。我想知道我是以正确的方式提交脚本，还是火花提交从来没有真正显示正在运行的应用程序。
localhost:8080或<master_ip>:8080不为我打开，但<master_ip>:
浏览 4提问于2016-12-01得票数 3
回答已采纳
我使用星星之火-SQL2.4.1版本和Kafka 0.10 v。
当我试图通过消费者来消费数据的时候。即使在将"auto.offset.reset“设置为”最新“之后，也会出现以下错误
org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Offsets out of range with no configured reset policy for partitions: {COMPANY_INBOUND-16=168}
at org.apache.kafka.clients.consumer.interna
浏览 5提问于2019-11-01得票数 1
我目前使用的是Spark中的Apache Beam2.29.0。我的管道使用来自Kafka的数据，我有一个自定义的KafkaConsumer，它是Beam通过调用ConsumerFactoryFn创建的。在运行期间，我需要在自定义Kafka消费者之间共享一段持久数据。在Spark中，这将非常简单，我将创建一个累加器变量，所有的执行器以及驱动程序都可以访问该变量。由于光束被设计为在多个平台上运行，Spark，Flink，Google Dataflow，它不提供此功能。有谁知道实现这一点的方法吗？
浏览 16提问于2021-09-10得票数 0
1回答
我循环遍历连接字符串数组，并在每个循环中提取一些信息并添加到列表中。现在，我想使用并行库来使其多线程，但我不确定该库是否保证写入列表是线程安全的，或者我是否需要使用锁定：
List<SomeType> list = new List<SomeType>();
settings.AsParallel().ForAll(setting =>
{
list.AddRange(GetSomeArrayofSomeType(setting)); /// DO I NEED TO DO LOCKING HERE???
})
浏览 0提问于2012-01-10得票数 4
回答已采纳
我正在学习火花和卡夫卡，并遇到了这个项目，它似乎有效地消耗了来自卡夫卡的信息。这个项目需要配置少量的kafka &动物园管理员属性，这是我挣扎的地方。我是说这个属性意味着什么zookeeper.broker.path？抱歉，如果这是个基本问题。
我在单个节点中配置了kafka，并具有以下属性，
broker.id=1
port=9093
log.dir=/tmp/kafka-logs-1
以及动物园管理员，
zookeeper.connect=localhost:2181/brokers
zookeeper.connection.timeout.ms=6000
如果我尝试用zookee
浏览 3提问于2015-10-07得票数 1
回答已采纳
我在一个简单的Kafka消费者程序上遇到了一些麻烦：
18/06/04 18:13:49 ERROR /log/log.txt: org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
at org.apache.kafka.clients.consumer.KafkaConsumer.<init>(KafkaConsumer
浏览 4提问于2018-06-04得票数 8
我正在使用Kafka运行一个结构化的流应用程序。我发现如果由于某种原因系统停机了几天...检查点变得陈旧，并且在Kafka中找不到与检查点对应的偏移量。我如何让Spark结构化流媒体应用选择最后一个可用的偏移量，并从那里开始。我尝试将偏移量重置设置为较早/最新，但系统崩溃，出现以下错误：
org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Offsets out of range with no configured reset policy for partitions: {MyTopic-574=6559828}
at
浏览 3提问于2020-04-19得票数 0
我试图访问一个卡夫卡主题使用火花流。我不认为我遗漏了任何依赖项或导入，但是当我试图运行代码时，如下所示：
public static void main(String[] args) {
String URL = "spark://localhost:7077";
SparkConf conf = new SparkConf().setAppName("Kafka-test").setMaster(URL);
JavaStreamingContext ssc = new JavaStreamingContext(conf, Durat
浏览 0提问于2018-02-09得票数 2
回答已采纳
我刚刚从Spark2.0.2升级到Spark2.1.0(下载Hadoop2.7及更高版本的预构建版本)。没有安装蜂巢。
在启动星体时，将在启动位置创建metastore_db/文件夹和derby.log文件，以及一堆警告日志(这些日志在上一个版本中没有打印)。
对调试日志的仔细检查表明，Spark2.1.0试图初始化一个HiveMetastoreConnection：
17/01/13 09:14:44 INFO HiveUtils: Initializing HiveMetastoreConnection version 1.2.1 using Spark classes.
Spark2.0
浏览 0提问于2017-01-13得票数 1
回答已采纳
1回答
我们的应用程序的hadoop集群安装了Spark1.5。但是由于特定的需求，我们使用2.0.2版本开发了spark job。当我向yarn提交作业时，我使用--jars命令覆盖集群中的spark库。但它仍然没有选择scala库jar。它抛出一个错误，说明
ApplicationMaster: User class threw exception: java.lang.NoSuchMethodError: scala.Predef$.ArrowAssoc(Ljava/lang/Object;)Ljava/lang/Object;
java.lang.NoSuchMethodError: scal
浏览 3提问于2016-12-16得票数 2
我是刚开始做运动的。我有需求，需要读取卡夫卡流和过滤数据和保存在数据库中。为此，我在KafkaItemReader中使用了spring批处理。当我在春季作业中运行start多个作业时，它给出了java.util.ConcurrentModificationException: KafkaConsumer对于多线程访问错误是不安全的。在这段时间里，它只运行最后一个作业。
这是spring批处理配置。
@Autowired
TaskExecutor taskExecutor;
@Autowired
JobRepository jobRepository;
浏览 2提问于2020-04-16得票数 1
回答已采纳
1回答
我有一个应用程序，它有两个线程正在运行。这些线程在它们之间共享一个KafkaConsumer实例。
KafkaConsumer是通过禁用自动提交配置创建的。一个线程将根据某些外部输入将主题分区分配给KafkaConsumer实例。另一个线程不断轮询主题，处理记录并在结束时提交偏移量。
现在，假设主题-分区分配在线程-1中发生了更改。我们如何确保Thread-2能够看到这些更改？也就是说，线程-2的KafkaConsumer将知道主题-分区已经在线程-1中更新。
将变量定义为易失性是否解决了这个问题？还是我们应该再做点什么？请提供您的解释与代码示例，因为我是新的多线程在Java。
浏览 4提问于2021-10-26得票数 2
回答已采纳
1回答
我使用的是带有Kafka的星火结构流，但是当我试图将流写到控制台时，我得到了错误：
Caused by: java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
这是我的代码：
def group_obs(obs_df):
obs = obs_df.select(f.col("obs.payload.after").alias("obs"))
filtered_obs_with_value = o
浏览 2提问于2018-04-25得票数 2
1回答
火花单元测试
、、
我的整个build.sbt是：
name := """sparktest"""
version := "1.0.0-SNAPSHOT"
scalaVersion := "2.11.8"
scalacOptions := Seq("-unchecked", "-deprecation", "-encoding", "utf8", "-Xexperimental")
parallelExecution in Test := fals
浏览 3提问于2016-12-16得票数 2
回答已采纳
1回答
build.sbt：
val sparkVersion = "2.1.1";
libraryDependencies += "org.apache.spark" %% "spark-core" % sparkVersion % "provided";
libraryDependencies += "org.apache.spark" %% "spark-sql" % sparkVersion % "provided";
libraryDependencies += "org
浏览 3提问于2017-06-09得票数 0
回答已采纳
我从数据库中读取数据，然后将数据放入一个JSON对象中，该对象也包含一个内部json对象。数据库中的一些数据返回为""，我希望删除该对象的键和值。 以下是数据的来源： {
"foo": "bar",
"bar": "",
"foo": {
"bar": "foo",
},
"bar": "foo"
}, 我希望数据看起来像这样： {
"foo": &
浏览 54提问于2019-04-10得票数 1
回答已采纳
在我的Android应用程序中，我使用这段代码，同时在地图上绘制一些路径点。
Iterator<Waypoint> iterator = waypoints.iterator();
while (iterator.hasNext()) {
Waypoint w = iterator.next();
}
但我发现了这个错误
致命例外: java.util.ConcurrentModificationException java.util.ArrayList$ArrayListIterator.next (ArrayList.java:573)
我是，而不是，直接在循环中
浏览 1提问于2018-03-10得票数 5
回答已采纳
我在AWS中创建了一个3个节点(1个主节点，2个工作节点)的Apache Spark集群。我可以从主机向集群提交作业，但是我不能让它远程工作。
/* SimpleApp.scala */
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
object SimpleApp {
def main(args: Array[String]) {
val logFile = "/usr/local/spark/RE
浏览 2提问于2017-02-05得票数 9
回答已采纳
我正在运行spark 2.0.2，并在火花独立集群上以cluster部署模式部署流作业。流作业工作正常，但是应用程序和驱动程序的stderr文件存在问题，这些文件是在SPARK_HOME的工作目录中创建的。由于流始终在运行，这些文件的大小只会增加，我不知道如何控制它。
我尝试了以下解决方案，尽管它们与手头的问题并不完全相关，但我仍然尝试过，但没有奏效：
谁能帮我限制创建这些文件的大小吗？
P.S:我尝试过在conf/spark-env.sh中添加下面一行并重新启动集群的解决方案，但是在运行流应用程序的情况下，它没有工作。
export SPARK_WORKER_OPTS=&
浏览 2提问于2017-02-09得票数 0
回答已采纳
我试着用Spark3.0.2阅读一个卡夫卡主题，用下面的库做一个火花外壳：
spark-sql-kafka-0-10_2.12-3.0.2.jarkafka-avro-serializer-6.2.0.jarkafka-clients-2.4.1.jarspark-streaming-kafka-0-10-assembly_2.12-3.0.2.jarspark-tags_2.12-3.0.2.jarspark-token-provider-kafka-0-10_2.12-3.0.2.jarcommons-pool2-2.6.2.jar
通过错误堆栈跟踪，我得到了以下输出：
---------
浏览 2提问于2021-07-21得票数 0
回答已采纳
我正在尝试从Kafka中读取，并使用Spark Structured Streaming写入Kudu Sink。下面是读写的代码。
我使用的是Spark 2.2.0。
val kafkaDataFrame = spark
.readStream
.format("org.apache.spark.sql.kafka010.KafkaSourceProvider")
.option("kafka.bootstrap.servers", kafkaReaderConfig.kafka_brokers)
.option("kafka.secur
浏览 0提问于2020-05-07得票数 0
我正在执行spark中的wordcount程序，并试图将结果存储在一个文本文件中。
我有一个scala脚本将单词计数为SparkWordCount.scala。我正在尝试从Spark控制台执行脚本，如下所示。
scala> :load /opt/spark-2.0.2-bin-hadoop2.7/bin/SparkWordCount.scala
Loading /opt/spark-2.0.2-bin-hadoop2.7/bin/SparkWordCount.scala...
import org.apache.spark.SparkContext
import org.apache.s
浏览 0提问于2017-10-10得票数 0
1回答
我正在Python上开发一个Spark脚本(使用Pyspark)。我有一个函数，它用一些字段返回一个Row，包括
timestamp=datetime.strptime(processed_data[1], DATI_REGEX)
processed_data1是一个有效的日期时间字符串。
编辑显示完整的代码：
DATI_REGEX = "%Y-%m-%dT%H:%M:%S"
class UserActivity(object):
def __init__(self, user, rows):
self.user = int(user)
浏览 5提问于2017-01-19得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
