我正在使用Spark XML库(HyukjinKwon:spark- XML :0.1.1-s_2.11)处理一个大的xml文件。XML处理失败，因为有几条记录出现了分析异常。我想继续处理文件，忽略这些记录。 我有以下处理xml的代码，我尝试了'DROPMALFORMED‘选项，但没有帮助。type mismatch: argument 2 requires integr
浏览 9提问于2019-04-05得票数 0
2回答
每天，大约有100~200行出现错误的可能性很大(状态列包含非空值)。从这些文件中，每天(因此是file_date的分区)，下游作业提取所有有效记录并将其发送以进行进一步处理，忽略这100-200条错误记录，等待接收到正确的映射文件。除了有效的状态记录之外，下游作业还应该尝试查看是否找到了错误记录的映射，如果存在，也应该进一步删除它(当然，在使用适当的映射和状态更新数据湖之后)。
最好的方法是什么？这是非常低效的，因为我们正在读取所有
浏览 42提问于2020-05-26得票数 1
1回答
我正在尝试使用Pyspark中包含25亿条记录的Hive表。我假设表中有一些格式错误的或其他“坏”数据。我使用的是spark版本1.6.2和python版本2.6.6：from pyspark.sql import我假设有一些Spark由于某种原因无法处理的记录。我想知道，在将Hive表作为pyspark dataframe加载后，是否有某种方法可以过滤它，这样只
浏览 1提问于2017-01-28得票数 1
1回答
我想用连衣裤。但是当我把它应用到我的代码中时，我得到了警告："drop_duplicates是不可调用的“。如果我将代码从"drop_duplicates“更改为"dropDuplicates"，则警告将消失。如何在不改变整个代码的情况下修复它？我认为drop_duplicates更像蟒蛇风格。我试着添加火花放电存根，但它确实有帮助。
浏览 5提问于2019-11-18得票数 2
1回答
在我们遗留DWH过程中，我们根据传统RDBMS中的rowid查找重复记录并跟踪重复记录。将仅返回对应最大记录的重复记录。即使当我们识别重复记录时，这也有助于识别/跟踪记录。
在pySpark中有没有类似的东西？如何在dwh到pyspark dwh转换项目中处理此问题？
浏览 11提问于2019-11-10得票数 0
回答已采纳
你好，我刚接触过Pyspark，当我将test.csv文件中的字符串列“道布”转换为Date格式时，我遇到了一个问题，在这个问题中，Pyspark将坏记录转换为空值。我知道pyspark在处理坏数据方面的方法，比如允许模式、FAILFAST模式、BadRecordPaths，它们在读取文件/输入时收集坏数据，但是在运行函数时有可能收集坏记录吗？因此，在运行这一行时，我希望收集NAME3和NAME4记录作为错误记录/
浏览 4提问于2022-06-01得票数 1
局部模式运行
from pyspark import SparkConf, SparkContext
rdd = sc.textFile('file:///myBigFile.xml
浏览 2提问于2019-01-31得票数 0
我有一个PySpark数据框，并且对于每个(批)记录，我想调用一个API。也就是说，假设我有100000k条记录，我想将条目批处理成1000条记录，然后调用一个API。我怎样才能用PySpark做到这一点？批处理的原因是因为API可能不会接受来自大数据系统的大量数据。 我最先想到的是LIMIT，但它不是“确定性的”。此外，它似乎是低效的？
浏览 27提问于2019-05-04得票数 3
回答已采纳
1回答
作为后台，代码所做的是处理一个销售数据集，以便找到与同一客户相关的以前的销售订单。但是，正如您将看到的，中间有很多业务逻辑，这可能会减慢速度。我正在考虑将其重构为PySpark工作，但在我这样做之前，我想知道这是不是完成这项工作的最好方法。 我将非常感谢这里的任何建议。 更多背景:每个循环大约需要10分钟才能完成。
浏览 18提问于2021-11-02得票数 0
回答已采纳
1回答
我正在在线监视数百个用户的活动，并试图获得最后5行的平均记录(忽略中间缺少的日期)。我使用Window.partitionBy在用户和rowsBetween之间进行分区，以获得最后5条记录，但不是为后面的每一行提供最后5条记录的平均值，而是将所有其他行都包含到平均值中。这就是我尝试过的：from pyspark.sql.window import Window
win = Window.partitionB
浏览 1提问于2021-02-23得票数 1
回答已采纳
1回答
所以我在一个SQL数据库中有一个表，我想使用Synapse (PySpark)来，添加新记录，覆盖现有记录()。但是，在PySpark中，我可以使用覆盖模式(这将删除我没有在迭代中推送的旧记录)，也可以使用附加模式(不会覆盖现有记录)。选项A：先加载旧记录，然后在PySpark中合并，然后对所有内容进行处理。缺点是我得先把整张桌子装上。
选项B：删除我将要覆盖的记录</
浏览 4提问于2022-03-23得票数 0
1回答
我正在使用内置Hadoop的PySpark 2.4.4处理一个大型数据集。数据集被分成多个压缩的CSV文件，我正在尝试使用PySpark的DataFrames报告各种列的聚合计数。在我要求PySpark使用以下命令将DataFrame写入磁盘之前，它一直处理得很好这会导致下面的警告淹没终端
WARN RowBasedKeyValueBatch我尝试使用spark.executor.memory属性增加执行器内存的大小；但是，错误
浏览 0提问于2020-02-23得票数 0
也许我应该显式地告诉hive.xml文件在哪里。我看到我有这份文件其他一些可能很重要的文件包括./etc/hive-hcatalog/conf.dist/proto-hive-site
浏览 0提问于2019-01-23得票数 1
我正在使用本地窗口，并试图用python上的以下代码加载XML文件，我遇到了这个错误，有人知道如何解决它吗？/201611339349202661_public.xml") 这就是错误 Py4JJavaError Traceback (most recent)
C:\SPARK_HOME\spark-2.2.0-bin-hadoop2.7\python\pyspark\sql\readwriter.py in load(self
浏览 87提问于2017-11-17得票数 4
回答已采纳
我正在尝试让读者从损坏的XML中恢复过来。在DOM (libxml2.readDoc)中使用libxml2.XML_PARSE_RECOVER选项是有效的，它可以从实体问题中恢复过来。它只是卡在一个永久循环中( reader.Read()返回-1)：import cStringIO
reader = libxml2.readerForDoc(DOC, &q
浏览 0提问于2010-10-06得票数 8
一旦在INFO级别开始记录，我就会在你的日志上看到一堆py4j.java_gateway:Received command c on object id p0。我怎么才能隐藏它呢？
浏览 1提问于2016-05-16得票数 18
xdr_predict File "/usr/bch/1.5.0/spark/python/lib/pyspark.zip/pyspark/ml/util.py", line 257, in load
File "/usr/bch/1.5.0/spark/python&
浏览 3提问于2021-12-03得票数 0
假设我在OracleJDK1.8(Build1.8.0_65-B17)上运行Spark1.6.0，在ipython笔记本会话中从下面的行开始：因此，我已经包含了databricks星火xml包()。接下来，我将针对pyspark运行以下代码：
dm
浏览 6提问于2016-01-26得票数 2
回答已采纳
1回答
我试图读取XML文件：df = spark.read.format('com.databricks.spark.xml').load('/path/to/my.xml')并获得以下错误：配置为：set jar_path = f'{SPARK_HOME}/jars/spark-xml</em
浏览 2提问于2020-09-18得票数 2
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
