Based on the introduction in Spark 3.0, https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html. It should be possible to set "kafka.group.id" to track the offset. For our use case, I want to avoid the potential data loss if the st...

我很难编写自己的spark scala代码以使用条件自连接来覆盖覆盖范围为空的行。 这是数据： +----+--------------+----------+--------+ | ID | date_in_days | coverage | values | +----+--------------+----------+--------+ | 1 | 2020-09-01 | | 0.128 | | 1 | 2020-09-03 | 0 | 0.358 ...

我对PySpark有点陌生，我希望实现它以加快脚本的运行时间，该脚本用于将美国的医师数据导入每个州的pdf文件和最终用于csv的csv文件数据库表。出于实用性考虑，由于我想扩展项目以查看不同国家的地区和省份，因此我希望能够使用PySpark进行分布式计算。 I can get processing done much more quickly, but I can't get my files to write back to the local system where I want to use my ...

我有一个Hive外部表作为TEXTFILE FORMAT hive> SHOW CRAETE TABLE customers; CREATE EXTERNAL TABLE `customers`( `id` int, `name` string, `age` int, `address` string, `salary` double) COMMENT 'Customer Details' ROW FORMAT SERDE 'org.apache.hado...

I am a beginner in Kafka and I am trying to read the data from a Kafka topic using spark in scala. 我的主要功能是： def main(args : Array[String]) : Unit = { val spark = SparkSession .builder() .appName("testKafka") .master("local[*]") .getOrCreat...

从类似日期（年）的列中，我得到了这种类型的值： +--------------------+------------+ | CUSTOMER_ID|yearSelected| +--------------------+------------+ |1 | 2010| |2 | 1992| |3 | 1996| |4 ...

我的代码很长。这段代码确实可以帮助我实现所需的功能，但是，我想对此进行一点整理。看起来我可以使用for循环来做到这一点，但是我是新手，所以对我来说一直是一个挑战。因此，我想知道是否有人可以使用for循环方法来帮助缩短我的代码！非常感谢，非常感谢您能提供的任何帮助！ df1=spark.read.format("parquet").load("dbfs:2020-01")\ .withColumnRenamed("count", "2020-01").sort('id').drop('date') df2=...

I am new to Scala and Spark. I am trying building a machine learning model. when i try to fit the model using training data using val model = pipeline.fit(trainingData) It throws error saying Task is not serializable. Please help me to resolve this issue. ...

我是pyspark / Apache Spark的新手。我需要从服务器上的数据库中获取多个表，每个表包含大约1.2亿行或更多行用于分析。我应该能够对数据进行计算。我在同时充当主服务器和从服务器的服务器上运行pyspark，并具有7.45G的RAM。 我已经安装了jdbc驱动程序，这是我使用的代码。 from pyspark.sql import SQLContext from pyspark import SparkContext sc = SparkContext.getOrCreate() sqlCon...

我编写了一个程序，该程序计算某些表值的一些协方差矩阵值，然后将这些协方差矩阵值添加到原始表中。该程序可以很好地计算整个表的协方差值，但是它给出了“当我将这些值加回到原始表时，编译失败。程序执行得很好，它还显示了数据帧的值，但是在不同的数据帧之间显示这无法编译错误。 以下是我的密码 def main(args: Array[String]): Unit = { val log = org.apache.log4j.LogManager.getLogger("Spark log") log.s...

This question already has answers here: (null) entry in command string exception in saveAsTextFile() on Pyspark ...

我通过指定分区数从文本文件创建RDD。但这给了我与指定分区不同数量的分区。 情况1 scala> val people = sc.textFile("file:///home/pvikash/data/test.txt", 1) people: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[50] at textFile at <console>:27 scala> people.getNumPartitions res36: Int = 1 s...

嗨，我正在尝试使用输入字符串计算给定问题中的最大值。 问题描述： 给定两个月的x和y，其中y> x，请找到从x月份到y月份增加最多推文数量的主题标签名称。我们已经在您的代码模板中编写了可从键盘读取x和y值的代码。忽略x和y之间月份中的推文，因此只需比较x月和y月份的推文数。报告主题标签名称，x和y个月中的推文数量。忽略在月份x或y中没有任何鸣叫的任何主题标签名称。您可以假设主题标签和月份的组合是唯一的。使用println将结果打印到终端输出。对于上面的小示例数据集，输出应为以下内容： 输入x = 20091...

我试图通过k8运行我的火花，我已经使用以下命令设置了RBAC： kubectl create serviceaccount spark kubectl create clusterrolebinding spark-role --clusterrole=edit --serviceaccount=default:spark --namespace=default 来自k8集群外部的Spark命令： bin/spark-submit --master k8s://https://<master_ip>:644...

val query = df.withColumn("value", col("value").cast(StringType)) .withColumn("value", from_json(col("value"), processor.Schema)) .select(unix_timestamp(col("timestamp")).alias("kafka_time"), col("value.*")) .filter(processor.filter) ...

我在PySpark中有一个dataFrame。我想将StringIndexer用于标签列，因此我将函数定义为： def indexer(column, dataframe): from pyspark.ml.feature import StringIndexer # Indexing the column stringIndexer = StringIndexer(inputCol=column, outputCol='categoryIndex') model = s...

我将使用以下两个数据帧来演示我的问题。 val datF1= Seq((1,"everlasting",1.39),(1,"game", 2.7),(1,"life",0.69),(1,"learning",0.69), (2,"living",1.38),(2,"worth",1.38),(2,"life",0.69),(3,"learning",0.69),(3,"never",1.38)).toDF("ID","token","value") datF1.s...

我正在尝试学习Spark，所以我是一个新手。 我有一个包含数千行的文件，每行的结构如下： LFPG;EDDW;00;E170;370;LFPG;EDDW;189930555;150907;1826;!!!!;AFR1724;AFR;0;AFR1724-LFPG-EDDW-20150907180000;N;0;;;245382;;;150907;1800;0;;X;;;;;;;;;;370;;0;20150907175700;AA45458743;;;;;NEXE;NEXE;;;;;2015090718000...

To understand this problem, imagine that a group of people, at time t1, are in a certain location L. You are interested in knowing where all those people were at time t0. 我在这里有一个示例代码来引导练习： row = Row("user_id", "start", "location_id") df = spark.sparkContex...

我有一个用例，可以向timestamp列中添加一些值并从中得出日期。我正在尝试使用from_unix和unix_timestamp函数。但是它以不同的格式显示了不同的值。 select from_unixtime('1577582583', 'YYYY-MM-dd'), from_unixtime('1577582583', 'yyyy-MM-dd'); from_unixtime(CAST(1577582583 AS BIGINT), YYYY-MM-dd) from_unixtime(CAST(...