我有拼图文件，假设文件名为abc/A。parquet和一些记录根据某些条件被过滤出来，并创建DF，我正在尝试用saveMode overwrite选项用结果过滤的DF覆盖文件，但抛出了以下异常：
用于覆盖filterDF.coalesce(1).write.mode("overwrite").parquet("file:/home/psub2/cls_parquet2/file:/home/psub7/abc/A.parquet")的命令
写入行时失败。在org.apache.spark.sql.execution.datasources.FileFormatW
浏览 16提问于2019-11-26得票数 0
回答已采纳
我在电子病历笔记本上执行了以下PySpark代码：
s3_path = "s3://bucket/key/file.csv"
df = spark.read.csv(s3_path, header=True)
df.repartition(1).write.mode("overwrite").csv(s3_path)
我得到了以下错误：
An error occurred while calling o166.csv.
: org.apache.spark.SparkException: Job aborted.
at org.apache.spark
浏览 12提问于2022-05-09得票数 0
回答已采纳
我已经将Spark配置为在两个节点上运行，并使用HDFS托管输入文件。我想将metrics.properties提供的所有统计数据文件转储到HDFS或每个节点的本地目录。
以下是我对metrics.properties的统计位置的配置：
*.sink.csv.directory=hdfs://ip:port/user/spark_stats/
我还尝试在每个节点中创建一个临时本地目录，并按如下方式配置metrics.properties：
*.sink.csv.directory=/tmp/spark_stats/
这两种方法都会产生如下错误：
16/03/02 15:41:49 WARN C
浏览 0提问于2016-03-03得票数 3
1回答
我有一个向ADLS Gen1目录写入(覆盖现有文件)的进程，然后另一个进程启动一个spark作业来读取最新覆盖的文件。
似乎大多数时候spark并没有读取最新的更新文件。在为第二个进程读取文件建立了一些延迟(30秒-60秒)之后，它似乎正在工作。
在不引入任何延迟的情况下解决此问题的最佳方法是什么？
感谢您的反馈。
浏览 0提问于2019-10-15得票数 0
我在同一个桶中存储了存储在S3上两个位置上的拼花文件：
path1: s3n://bucket/a/
path2: s3n://bucket/b/
数据具有相同的结构。我希望从第一个位置读取文件，并使用spark将它们聚合到第二个位置。下面是代码片段：
val df1 = sql.read.parquet(path1)
val df2 = sql.read.parquet(path2)
val df = df1.unionAll(df2)
df.write.mode(SaveMode.Overwrite).parquet(path1)
当我运行这段代码时，我会得到以下异常：
java.io
浏览 0提问于2016-05-16得票数 0
回答已采纳
2回答
我有一个DSE集群，集群中的每个节点都在运行spark和Cassandra。
当我从Cassandra加载数据到spark rdd并在Rdd上执行一些操作时，我知道数据将分布到多个节点中。在我的例子中，我想把这些rdds从每个节点直接写到它的本地Cassandra dB表中，有没有办法做到这一点。
如果我执行普通的rdd收集，来自spark节点的所有数据将被合并，并返回到带有驱动程序的节点。我不希望发生这种情况，因为从节点返回到驱动程序节点的数据可能需要很长时间，我希望数据直接保存到本地节点，以避免数据跨spark节点移动。
浏览 3提问于2020-10-20得票数 1
我已经用一个列(例如，id)对数据进行了分区，并将这个数据集保存在某个位置。有时，我会得到一个结构相同的较小的增量数据集，实际上，我必须在id的基础上重新插入我现有的数据，并在date列中决定哪条记录是最新的。(我不把它写在同一个地方，我把整个新的blob都放在别的地方了。)
我一直在做这件事，有两种方法--要么在窗口中分组，要么使用最高的date行。或者通过dropDuplicates，依靠事实，我的数据是有序的。(我宁愿使用前者，但我一直在尝试各种方法。)
一个大问题是，每个id组都是不可忽略的(几千兆字节)，所以我希望Spark (与n员工一起)能够理解，既然我正在读取id-partit
浏览 0提问于2018-03-26得票数 0
1回答
我使用文件作为Spark流，我想计算流中的单词，但是应用程序什么也不打印，这是我的代码。我在Cloudera环境中使用Scala
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext
object TwitterHashtagStreaming {
def main(args: Array[String]) : Unit = {
val conf = new SparkConf().setAp
浏览 4提问于2017-09-26得票数 0
回答已采纳
我有一个火花作业，它在第一次运行时在5分钟内运行，然后需要几分钟..more，而在随后的运行中需要20-30分钟。我正在读取一个拼花文件一次，然后创建数据格式并以.json格式编写。在代码中的任何地方，我都没有使用缓存()、持久化()或未持久化()。这是本地实例。有什么问题吗？
配置参数
val spark = SparkSession
.builder()
.appName("example")
.config("spark.sql.warehouse.dir", warehouseLocation)
.config("spark
浏览 2提问于2017-07-28得票数 0
1回答
在解析pyspark中不一致的数据类型时，我遇到了一个问题。如下面的示例文件所示，SA键总是包含一个字典，但有时它可以显示为string值。当我尝试获取列SA.SM.Name时，会得到如下所示的异常。
对于SA.SM.Name列，如何将null设置为除了JSONs之外的值的pyspark/hive。有人能帮帮我吗？
我试图转换成不同的数据类型，但是没有工作，或者我可能做错了什么。
输入文件内容: mypath
{"id":1，"SA":{"SM"：{"Name"："John"，"Email"：“Jo
浏览 6提问于2020-10-10得票数 0
1回答
我正在用下面这行代码读取我的主CSV文件
master_df = spark.read.format('csv').option("encoding", "cp1252").option("header", True).load(master_csv_path).cache()
然后我读取一些其他CSV文件并将它们合并到master中，然后我想覆盖master CSV文件
master_df.coalesce(1).write.format("csv").mode("overwrite").opti
浏览 0提问于2020-09-29得票数 0
我正在尝试将一个拼花文件写入HDFS，然后将其复制到s3。
我用齐柏林飞艇( Zeppelin )编写了代码，效果很好。没有任何问题，它将文件添加到s3文件路径。
var outputFolder = "buckent_name/path"
println("\n ---- TASK 1 ----- \n writing with path " + outputFolder)
wholeParquetFile
.withColumn("date_col", to_date(col("timestamp"), &
浏览 11提问于2022-05-17得票数 -1
回答已采纳
1回答
我是Apache的新手，我正在读取HDFS目录中的文件，然后根据条件进行过滤和排序。
我在hdfs目录中有两个文件，第一个文件包含以下数据
Name:xxxx,currenttime:[timestamp],urlvisited:[url]
包含以下信息的第二个文件
Name:xxxx,currenttime :[timestamp],downloadfilename:[filename]
首先，我根据名称过滤数据，然后使用逗号分割数据，然后使用当前时间字段对数据进行排序。
到目前为止我已经试过了
import java.text.ParseException;
import java.uti
浏览 1提问于2015-04-13得票数 0
最近，我开始研究Talend Bigdata。我已经编写了一些正在正确执行的作业。每次运行中要处理的数据不到1 gb，但它是.gz文件。作业正在成功运行，但我经常遇到下面的错误，尽管当重新处理作业时没有任何更改，那么它是成功的。Executor内存和内存开销按预期设置。GC收集器也设置为G1GC。一个能帮到我的主意。如果你有更多的信息，请让我知道。
ERROR executor.CoarseGrainedExecutorBackend: Driver 131.116.127.73:45245 disassociated! Shutting down.
[ERROR]: org.apache.
浏览 0提问于2018-04-16得票数 1
1回答
星火中的集散水槽流
、、、
我正在研究在我们的设置中做一些“实时”日志处理的能力，我有一个关于如何进行的问题。
因此，当前的设置(或我们打算这样做)如下：
服务器A通过Rsyslog生成日志文件到每个客户的文件夹。
服务器B通过Rsyslog生成日志文件到每个客户的文件夹。
服务器A和B在每个客户的文件夹中最多生成15个日志文件(每个客户1个)，结构如下：
/var/log/CUSTOMER/logfile.log
在服务器C上，我们有一个侦听来自服务器A和服务器B的Rsyslog tcp消息的Flume接收器。目前，为了测试，我们只为一个客户提供了一个水槽接收器，但我认为每个客户需要一个水槽接收器。
然
浏览 1提问于2015-04-09得票数 1
1回答
AWS胶-书写文件需要很长时间
、、、、
你好，我在AWS中有一份ETL工作，需要很长时间才能编写。它从S3读取数据并执行几个转换(所有转换都没有在下面列出，但这些转换似乎不是问题所在)，然后最后将数据帧写入S3。然而，这种写作操作似乎需要很长时间。即使在使用10 worker(工人类型G.1X)时，对于一个大约20 MB的文件，也可以使用Approx 30分钟。我使用了print语句来查看什么需要时间，这似乎是将文件写入S3的最后一次操作。在使用相同的设置之前，我还没有遇到过这个问题。
我正在使用Glue version 3.0、3和Spark版本3.1。
源文件中的文件数量几乎为50000个文件，分布在许多文件夹中，每天都会自动生
浏览 6提问于2022-03-29得票数 2
1回答
版本: emr-5.33.1，pysplk2.4.7
尝试迭代读取数据子集，对其进行转换，然后保存它们以保存桶。路径如下所示：{bucket_uri}/folder1/date=20220101/
当没有这样的date=20220101分区文件夹时，写入是可能的，但是它输出。
pyspark.sql.utils.AnalysisException: 'path ... already exists'
我的代码如下所示：
output_path = 'bucket_uri/folder1/date=20220101'
for i in range(0, 100,
浏览 10提问于2022-11-23得票数 0
回答已采纳
1回答
上下文:我有一个使用SQl查询的数据文件。在这个查询中，我使用spark上的熊猫保存到一个数据文件中。现在，经过一些转换后，我想将这个新的dataframe保存在给定数据库的新表上。
示例：
spark = SparkSession.builder.appName('transformation').getOrCreate()
df_final = spark.sql("SELECT * FROM table")
df_final = ps.DataFrame(df_final)
## Write Frame out as Table
spark_df_fina
浏览 13提问于2022-11-18得票数 0
回答已采纳
我有一个简单的Spark作业，映射，计算和写入Oracle DB的结果。我在将结果写入数据库时遇到了问题。
在按键减少结果之后，我将调用foreachPartition操作来建立连接并将结果写入DB。如果我将并行化设置为1，它会工作得很好。但是，当我将reducer的并行化更改为2或更大时，它只写入部分结果。当我检查日志文件时，我看到以下错误：
java.sql.BatchUpdateException: ORA-00060: deadlock detected while waiting for resource
我该如何解决这个问题？
浏览 2提问于2015-06-05得票数 1
1回答
正在尝试从hdfs加载avro。我有大约1000部分avro文件在一个目录中。我正在用这个来读它们-
val df = sqlContext.read.format("com.databricks.spark.avro").load("path/to/avro/dir") df.select("QUERY").take(50).foreach(println)
如果我在path中只传递了1或2个avro文件，它就会起作用。但是如果我用400+文件传递一个目录，我会得到这个错误。每个avro大约是300mb。
org.apache.avro.Avr
浏览 2提问于2016-01-31得票数 0
1回答
我使用的是容器，它基于容器。
我如何解决这个异常：
org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: file:/README.md
Main.java
context = new SparkContext(
new SparkConf()
.setAppName("Test App")
.setMaster("spark://s
浏览 0提问于2018-08-08得票数 0
我有一个包含5个节点的spark独立集群。所有节点都通过nfs挂载了相同的卷。这些挂载中的文件具有特定的linux文件权限。
当我以用户x(在所有节点上都可用，并且在所有节点上具有相同的uid)的身份spark提交作业时，我希望spark executors也以用户x的身份运行，以便作业只能访问用户x有权访问的文件。
我没有Kerberos，也没有HDFS。
在这个设置中这是可能的吗？使用纱线会有帮助吗？
浏览 0提问于2020-02-22得票数 1
1回答
我有大约500-700 csv文件的日志加载每周。文件名是每周随机生成的。这些文件存储在google云存储上，pyspark在google dataproc上运行，有几个工作节点。
我已经创建了一个pyspark文件，它基本上只是迭代文件并按原样加载到bigquery中。我以为我会在bigquery表上看到增量加载，但似乎pyspark会将整个gz输入文件合并到dataframe中，然后立即处理它。所以我在大约90分钟后得到了结果。
我如何编写pyspark代码，以便每次读取文件时，都会将其写入bigquery。总体时间可能不会有太大变化，但至少我如何才能看到部分写入bigquery的数据？
浏览 22提问于2021-10-10得票数 1
我将作业提交到运行在Yarn上的星火集群--有时通过相对缓慢的连接提交。为了避免为每个作业上传156 job火花组装文件，我将配置选项spark.yarn.jar设置为HDFS上的文件。但是，这并不能避免上传，而是从HDFS目录获取程序集文件，并将其复制到应用程序目录：
$:~/spark-1.4.0-bin-hadoop2.6$ bin/spark-submit --class MyClass --master yarn-cluster --conf spark.yarn.jar=hdfs://node-00b/user/spark/share/lib/spark-assembly.jar
浏览 2提问于2015-07-06得票数 1
回答已采纳
据推测，Apache支持Facebook的Zstandard压缩算法，从Spark2.3.0 ()开始，但我无法真正读取Zstandard压缩文件：
$ spark-shell
...
// Short name throws an exception
scala> val events = spark.read.option("compression", "zstd").json("data.zst")
java.lang.IllegalArgumentException: Codec [zstd] is not available.
浏览 7提问于2018-06-15得票数 8
回答已采纳
我有一个严格的I/O限制(Java)光束管道，在Google Cloud Dataflow上，我使用数据流光束选项"setNumberOfWorkerHarnessThreads( 16 );“在每个虚拟CPU上运行16个线程。我正在尝试将相同的管道移植到Spark上运行，但我在Spark上找不到相同的选项。我尝试过自己的线程化，但这似乎在SparkRunner上造成了问题，因为DoFn的ProcessElement部分会返回，但稍后在线程完成时会调用ProcessContext的输出。(我得到了奇怪的ConcurrentModificationExceptions，堆栈跟踪是bea
浏览 12提问于2021-05-05得票数 0
目前正在为10 kb文件生成两个avro文件，如果我对我的实际文件(30MB+)遵循相同的操作，我将生成n个文件。
因此，即使源文件很大，也只需要一个或两个.avro文件的解决方案。
此外，是否有任何方法可以避免手动声明列名。
目前的方法..。
导入org.apache.spark.sql.types.{StructType，StructField，StringType}
//手动模式声明'co‘和'id’列名和类型val customSchema =StructType( StructField("ind"，StringType，true)，StructF
浏览 0提问于2018-11-06得票数 1
回答已采纳
我正尝试在spark2中运行一个批处理作业，它接受一个巨大的列表作为输入，并在该列表上迭代以执行处理。该程序对于列表的大约8000个记录执行良好，然后中断，给出异常：
WARN Lost task 0.0 in stage 421079.0 (TID 996338, acusnldlenhww4.cloudapp.net, executor 1): java.io.FileNotFoundException: /data/1/hadoop/yarn/local/usercache/A2159537-MSP01/appcache/application_1497532405817_0072/b
浏览 0提问于2017-06-22得票数 0
1回答
文件写入的火花分区非常慢。
、、、、
当使用Spark向HDFS写入文件时，当不使用分区时，这是相当快的。相反，当我使用分区写入文件时，写入延迟增加了24因子。
对于同一个文件，没有分区的写入大约需要600 For。用Id编写分区(将生成准确的1.000个分区，因为文件中有1.000个Id)，大约需要14秒。
你们中的一些人是否有过编写分区文件需要很长时间的相同经验？造成这种情况的根本原因是什么，也许火花需要为每个分区创建1.000个文件夹和文件？你知道如何加速这件事吗？
val myRdd = streamedRdd.map { case ((id, metric, time), value) => Record(id,
浏览 4提问于2016-04-01得票数 3
一般来说，如何在不输出文件的情况下从spark作业中提取结果？
比方说，我有一个scala程序，它创建一个spark launcher作业。
我可以用Futures完成这个spark任务。一旦spark作业完成，我如何才能得到结果呢？(我之前的一种方法是让spark作业将结果写到一个文件中，然后外部scala程序读取该文件。<-我想避免这种策略)
浏览 1提问于2015-12-23得票数 0
我需要用新数据对旧数据执行update insert (Upsert)。
伪代码：
old_data = spark.read.parquet('s3://bucket/old_data/')
new_data= spark.read.parquet('s3://bucket/new_data/')
common_records = old_data.join(new_data,on=opk,how="inner")
non_match_records = old_data.join(new_data,on=opk,how="left_
浏览 0提问于2021-09-24得票数 1
2回答
我正在使用Spark2.3，我需要将一个保存到一个csv文件中，我正在寻找一种更好的方法来实现它。通过查看相关/类似的问题，我发现了，但我需要一个更具体的：
如果DataFrame太大，我如何避免使用Pandas？因为我使用了toCSV()函数(下面的代码)，它产生了：
内存不足错误(无法分配内存)。
使用文件I/O直接写入csv是更好的方式吗？它能保护分离器吗？
使用df.coalesce(1).write.option("header", "true").csv('mycsv.csv')将导致在每个文件中写入标头，当文件合并时，头部将
浏览 2提问于2018-06-06得票数 1
4回答
我正在将数据集加载到DynamicFrame中，执行转换，然后将其写回S3：
datasink = glueContext.write_dynamic_frame.from_options(
frame = someDateFrame,
connection_type = "s3",
connection_options = {
"path": "s3://the-bucket/some-data-set"
},
format = "parquet"
)
结果是12个拼
浏览 2提问于2019-10-06得票数 2
亲爱的，
我正在考虑如何在给定的通用数据中心体系结构中正确地使用Streamset：
我有几种数据类型(csv，tsv，json，来自IOT的二进制文件)，需要由CDC捕获并以as-is格式保存到Kafka主题中，然后以-is的形式进入HDFS数据湖。
然后，另一个Streamset管道将使用这个Kafka主题，并将其转换为通用格式(取决于数据类型)为JSON，并执行验证、掩蔽、元数据等操作，并保存到另一个Kafka主题。
相同的JSON消息将以Avro格式保存到HDFS数据湖中进行批处理。
然后，假设JSON数据已经准备就绪，并且可以进一步丰富其他数据，以便进行可伸缩的复杂转
浏览 0提问于2018-06-11得票数 1
我有一个spark应用程序，它将处理后的输出转储到S3存储桶中。我的输出数据帧在pyspark中的分区数是300。我有列"id“(大约250个唯一值)，我用它来用partitionBy写文件。因此，我得到了大约75K个文件，总和为11 to。
在我的spark应用程序中，将此数据写入S3的所有阶段都已完成，但从_temp文件夹中移动文件仍在进行中。这是非常缓慢的，按照目前的速度，这将需要一天以上的时间。这里的根本原因是什么?如何优化它？
浏览 1提问于2020-06-05得票数 0
我对Hive的理解是，它提供了一种使用SQL命令查询HDFS的方法。
好吧，但是还有Spark。Spark拥有所有的RDD类方法，这些方法完全有能力，但我更喜欢使用SQL。
输入Spark SQL。既然我可以使用Spark SQL通过SQL查询我的数据库，那么为什么Hive会进入这个画面呢？
医生说：
enableHiveSupport():
Enables Hive support, including connectivity to a persistent Hive metastore, support for Hive serdes, and Hive user-defined fun
浏览 1提问于2018-08-16得票数 0
1回答
我正在从container1中检索两个文件，转换它们，并在写入到Azure相同存储帐户中的container2之前进行合并。我正在安装container1，在编写之前打开和安装countainer2。
我写拼花的代码
spark.conf.set("spark.sql.sources.partitionOverwriteMode","dynamic")
df_spark.coalesce(1).write.option("header",True) \
.partitionBy('ZMTART') \
浏览 8提问于2022-11-22得票数 0
我有一个包含列id和几个字符串列的.csv文件。我想按id分组，然后将string_column1中的所有值写入一个文本文件(每个值在一个新行上)。最后，我希望文本文件的名称是“allstring”+id。我在Java中使用Apache Spark。
我尝试过使用groupBy("id").agg(collect_list("string_column1"))，但是我得到了“方法collect_list(String)对于类型Main是未定义的”。
我不知道如何使用id列中的不同值来命名文本文件。
import org.apache.log4j.Level;
i
浏览 2提问于2019-11-02得票数 0
我想把数据格式写成.delta格式，但是得到了“AnalysisExcpetion”
代码：
df = spark.read.format("csv").option("header", True).option("inferSchema",True).load('dbfs:/mnt/raw/Sales.csv')
df.write.format("delta").mode("overwrite").save('/mnt/bronze/sales/')
Exception:
--
浏览 8提问于2022-07-12得票数 0
如果我的卡夫卡主题收到这样的记录
CHANNEL | VIEWERS | .....
ABC | 100 | .....
CBS | 200 | .....
我有Spark结构化的流媒体代码来读取和处理Kafka记录如下：
val spark = SparkSession
.builder
.appName("TestPartition")
.master("local[*]")
.getOrCreate()
import spark.implicits._
浏览 0提问于2018-04-20得票数 5
1回答
我用的是电火花
我读取了一个libsvm文件，转换它，然后再保存它。
我将每个数据行保存为具有稀疏数据的对象。
我尝试使用MLUtils.saveaslibsvm，而不是使用MLUtils.loadlibsvm读取文件，并得到以下错误
ValueError:无法将字符串转换为浮动：[
在org.apache.spark.api.python.PythonRunner$$anon$1.read(PythonRDD.scala:193) at org.apache.spark.api.python.PythonRunner$$anon$1.(PythonRDD.scala:234) at org
浏览 2提问于2017-08-10得票数 0
回答已采纳
在火花提交时，我的印象是，如果我的应用程序依赖于其他.py文件，那么我必须使用py-file选项分发它们(参见)。我认为这意味着任何文件都必须使用py文件来声明，但是下面的操作很好.两个.py文件：
spark_submit_test_lib.py
def do_sum(sc) :
data = [1, 2, 3, 4, 5]
distData = sc.parallelize(data)
return distData.sum()
和spark_submit_test.py：
from pyspark import SparkContext, SparkConf
from spa
浏览 2提问于2016-10-03得票数 0
回答已采纳
有没有一个配置用来控制像在Presto中那样使用INSERT或CREATE TABLE写入的文件数？寻找与Spark对应的spark.sql.shuffle.partitions = 1相似或相同的东西。 我希望减少与插入小文件的数量，以避免与上述火花配置在Spark中额外的ETL。这个是可能的吗？我还没有在Presto文档中找到类似的东西。
浏览 83提问于2019-12-27得票数 1
我正在开发集群上的朱庇特笔记本。当您使用笔记本时，它会在每个单元格的执行过程中给出输出。
我必须在集群上提交一个PySpark作业。作业接受一个.py文件作为输入。截图附后
当我输入一个.ipynb文件的路径时，它会出现以下错误。
线程" main“中的file:/tmp/job-e4811479-cell2cell1/Customer_Churn_(Cell2Cell)_Parallel.ipynb异常:无法从JAR、org.apache.spark.deploy.SparkSubmitArguments.error(SparkSubmitArguments.scala:6
浏览 3提问于2020-06-06得票数 1
回答已采纳
def synapsedump(targetmount,targetfolder,table,df):
dbutils.fs.rm("/mnt/tmp", recurse=True)
df.createOrReplaceTempView(table)
spark.conf.set("spark.databricks.sqldw.writeSemantics", "copy")
schema = "Amazon"
schematable = schema + "." + table
df =
浏览 39提问于2021-08-03得票数 0
如何在spark中将文件保存在hdfs服务器上 以下是我的代码 csv_file.write.mode('overwrite').parquet('hdfs:///data/ifckp/docs/csv_files/cs') 下面是我的错误： Py4JJavaError: An error occurred while calling o3621.parquet.
: org.apache.hadoop.security.AccessControlException: Permission denied: user=zeppelin, access=WRITE
浏览 25提问于2020-02-01得票数 0
我使用下面的代码提交了我的jar文件：
spark-submit \
--class Hello \
--master spark://master:7077 \
--num-executors 6 \
--conf spark.executor.extraJavaOptions="-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+UseCompressedOops" \
first.jar
您可以看到，我按照官方文档中的建议添加了-verbose:gc -XX:+PrintGCDetails -XX:+
浏览 16提问于2016-09-23得票数 1
1回答
我有以下包含数据的数据框
+---------------------------+-------+
|sport |value |
+---------------------------+-------+
|table tennis |12 |
+---------------------------+-------+
我想把这个数据帧写到一个现有的csv文件中。我的代码如下
val existingSparkSession = SparkSession.builder().getOrCreate()
浏览 20提问于2021-02-25得票数 0
1回答
我的目标是使用Spark结构化流将转换后的数据从MongoDB集合写入Neo4j。根据Neo4j文档，这在"“版本4.1.2中是可能的。
到目前为止，批处理查询工作正常。但是，通过下面的示例，我遇到了一条错误消息：
spark-shell --packages org.mongodb.spark:mongo-spark-connector:10.0.2,org.neo4j:neo4j-connector-apache-spark_2.12:4.1.2_for_spark_3
val dfTxn = spark.readStream.format("mongodb")
.
浏览 17提问于2022-06-27得票数 0
我想把分散在s3上的多个分区层的小块文件压缩成更大的文件，只有一个分区才能返回到s3。
因此，在本例中，我有3个分区层(part1、part2、part3)。我想把这些数据写回只由part2分区的
在我的第一次跑步中我使用了：
df = spark.read
.option("basePath", "s3://some_bucket/base/location/in/s3/")
.parquet("s3://some_bucket/base/location/in/s3/part1=*/part2=*/part3=*/")
df.write.p
浏览 2提问于2018-04-11得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
