Spark在读取CSV时跳过错误记录
Spark是一个开源的大数据处理框架，可以在分布式环境中进行高效的数据处理和分析。它提供了丰富的API和工具，支持多种数据源和数据格式。
在读取CSV文件时，Spark提供了一种跳过错误记录的机制。当CSV文件中存在格式错误或者不符合预期的记录时，可以通过设置相应的参数来跳过这些错误记录，继续读取有效的数据。
具体来说，可以使用Spark的CSV数据源库（如spark-csv）来读取CSV文件。在读取时，可以通过设置mode参数为PERMISSIVE来启用跳过错误记录的功能。这样，Spark会尝试解析所有记录，将解析成功的记录作为有效数据返回，而将解析失败的记录标记为错误。
除了mode参数，还可以通过其他参数来进一步控制错误记录的处理方式。例如，可以设置columnNameOfCorruptRecord参数来指定一个列名，将解析失败的记录放入该列中；还可以设置badRecordsPath参数来指定一个路径，将解析失败的记录保存到该路径下的文件中，以便后续分析和处理。
Spark的跳过错误记录机制可以帮助用户在处理大规模数据时快速定位和处理错误，提高数据处理的鲁棒性和效率。
腾讯云提供了一系列与Spark相关的产品和服务，可以帮助用户在云上快速搭建和运行Spark集群，如腾讯云EMR（Elastic MapReduce）服务。EMR是一种弹性的大数据处理服务，支持Spark等多种大数据框架，提供了简单易用的界面和管理工具，帮助用户快速部署和管理Spark集群。您可以通过访问腾讯云EMR的官方网站（https://cloud.tencent.com/product/emr）了解更多相关信息。
相关·内容
2回答
在Spark中，在CSV和JSON上查询DataFrames是否有性能上的差异
amazon-web-services、amazon-s3、apache-spark、apache-spark-sql
我在AWS S3上有一个CSV文件和一个JSON文件(每个文件都有500万行/记录)。它们包含完全相同的数据，只是格式不同。
在有10个任务节点的EMR集群上，我启动了Spark (10个执行器，40个执行器核心)，并创建了两个DataFrames:一个针对CSV，一个针对JSON。
对构建在JSON文件上的DF的查询运行速度比对CSV构建的DF的类似查询快2-3倍。我没有发现任何关于跨存储文件格式的性能差异的信息。
有没有人知道为什么对JSON上的DF的查询比CSV上的DF运行得更快？
在下创建数据仓库的代码
针对JSON文件创建DF：
val hc_json = new org.apach
浏览 5提问于2015-11-03得票数 3
1回答
PySpark XML处理-忽略错误记录
pyspark、azure-databricks
我正在使用Spark XML库(HyukjinKwon:spark- XML :0.1.1-s_2.11)处理一个大的xml文件。XML处理失败，因为有几条记录出现了分析异常。我想继续处理文件，忽略这些记录。 我有以下处理xml的代码，我尝试了'DROPMALFORMED‘选项，但没有帮助。 df = (spark.read.format("xml")
.option("rootTag","Articles")
.option("rowTag", "Article")
浏览 9提问于2019-04-05得票数 0
3回答
使用架构、头检查和存储损坏的记录读取csv
python、csv、apache-spark、pyspark
我正在尝试使用pyspark读取器，其标准如下：
根据模式中的数据类型读取csv
检查标题和架构匹配中的列名。
在新字段中存储已损坏的记录
这是我尝试过的。
file: ab.csv
------
a,b
1,2
3,four
from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
DDL = "a INTEGER, b INTEGER"
df = spark.read.csv('ab.csv', header=True, schema=
浏览 0提问于2019-03-07得票数 10
1回答
为什么Spark应用程序将包含多个csv文件的DataFrame保存到S3存储桶
amazon-web-services、apache-spark、amazon-s3、amazon-emr
嗨，我是Spark和Amazon EMR集群的新手。 我试图编写一个可以在Amazon EMR集群上运行的演示spark应用程序。当代码在Zeppelin notebook上运行时，它会返回输出，我认为输出会保存为Amazon EMR集群上的单个文件，如下所示： %pyspark
spark.conf.set('spark.sql.repl.eagerEval.enabled', True)
if type(df.c) == type(upper(df.c)) == type(df.c.isNull()):
df_new = df.withColumn('up
浏览 31提问于2021-04-16得票数 0
回答已采纳
3回答
EMR上的火花究竟是如何从S3中读取的？
amazon-web-services、apache-spark、amazon-s3、amazon-emr
关于将s3上的文件读入带有Spark的EMR集群中的实际机制，只需问几个简单的问题：
spark.read.format("com.databricks.spark.csv").load("s3://my/dataset/").where($"state" === "WA")是否将整个数据集传送到EMR集群的本地HDFS，然后执行筛选？还是在将数据集带入群集时对记录进行筛选？还是两者都不起作用？如果是这样的话，到底发生了什么？
官方文档缺乏对正在发生的事情的解释(或者如果它有解释，我就找不到它)。有人能用这样的解释来解释，
浏览 0提问于2018-01-20得票数 1
回答已采纳
2回答
EMR引导失败
amazon-web-services、apache-spark、pyspark、amazon-emr
我正尝试在AWS EMR中运行以下引导程序脚本，但它失败了，没有提供错误消息。我已经为应用程序设置了EMR集群: Spark、Hive、Ganglia和Livy。
!/bin/bash
sudo pip-3.6安装-U \ matplotlib \ pandas \ spark-nlp
我使用的是EMR版本6。之前我们只使用"pip“而不是"pip- 3.6”，但它仍然出错，因此我们决定尝试使用"pip-3.6“，因为我们假设EMR 6有python 3.6。
请让我知道这可能有什么问题。
谢谢!
浏览 12提问于2020-05-21得票数 0
2回答
火花上的Scala数据分析
scala、csv、apache-spark、dataframe、rdd
我是Scala新手，我必须使用Scala和Spark的SQL、Mllib和GraphX来对巨大的数据集进行一些分析。我想做的分析是：
客户生命周期价值(CLV)
中心性度量(度、特征向量、边缘间度、贴近度)数据位于Hadoop集群中的CSV文件(60 in (3年跨国数据))中。
我的问题是访问数据和执行上述计算的最佳方法？
我是否应该将CSV文件中的数据加载到dataframe中并进行数据处理？或
我是否应该从CSV文件中加载数据并将其转换为RDD，然后处理RDD？或
是否有其他方法来访问数据和进行分析？
提前谢谢你的帮助。
浏览 2提问于2016-09-25得票数 3
3回答
如何将csv文件加载到RStudio上的SparkR中？
r、apache-spark、apache-spark-sql、sparkr
如何将csv文件加载到RStudio上的SparkR中？以下是我在RStudio上运行SparkR所必须执行的步骤。我已经用read.df读取了.csv，不知道还能怎么写。不确定是否将此步骤视为创建RDDs。
#Set sys environment variables
Sys.setenv(SPARK_HOME = "C:/Users/Desktop/spark/spark-1.4.1-bin-hadoop2.6")
.libPaths(c(file.path(Sys.getenv("SPARK_HOME"), "R", "lib
浏览 0提问于2015-10-01得票数 5
1回答
在S3中合并大量小csv文件(不同列大小)的最佳方法
python、csv、apache-spark、amazon-s3
我有一大堆小的csv文件(每个大约15kb )。这些csv文件具有不同的标头。我想将它们合并到一个包含所有列的csv文件中。最终，合并后的csv文件可能会超过100 be。合并文件时，如果列不存在，则添加此列并填充null。我从s3下载这些csv文件到EMR集群，然后使用Spark代码合并这些csv文件，并尝试使用python代码合并它们。从我所做的实验来看，Python代码比Spark运行得更好，但不是idea。有没有更好的主意？谢谢。
浏览 3提问于2018-06-07得票数 1
2回答
Amazon EMR集群上的spark错误
apache-spark、amazon-emr、spark-csv
我试图运行一个EMR集群与一个简单的火花步骤执行，我遇到了一个错误，我无法解决。当我在Eclipse中本地运行它时，这个程序会工作，但是当我在EMR集群上运行它时，这个程序就不能工作了。该程序只是试图将S3上的CSV文件转换为Parquet格式。
当我在EMR中运行时，我会得到以下错误：
原因是:解析输入的com.univocity.parsers.common.TextParsingException:长度(1000001)超过解析器设置中定义的最大字符数(1000000)。已识别的已解析内容中的行分隔符。这可能是错误的原因。解析器设置中的行分隔符设置为“\n”。分析内容：
我没有超
浏览 6提问于2017-01-24得票数 0
1回答
用于大型csv文件的sparklyr
r、apache-spark、sparklyr
我正在尝试使用sparklyr加载一个包含一百万行和1000列的数据集。我在工作中的一个非常大的集群上运行Spark。尽管如此，数据的规模似乎太大了。我尝试了两种不同的方法：
这是数据集：(train_numeric.csv)
1) -将路径放入hdfs - spark_read_csv(spark_context，.csv )
2) -将csv文件作为常规R dataframe读取- spark_frame<-copy_to(sc，R-dataframe)
这两种方法在数据集的子集上都工作得很好，但当我尝试读取整个数据集时，这两种方法都会失败。
有没有人知道适用于大型数据集的方法？
浏览 2提问于2017-05-30得票数 0
1回答
从RDS到S3 -数据转换
java、amazon-web-services、apache-spark、amazon-emr、data-transfer
我的/ oracle中有大约30个表(还没有决定是oracle还是postgres )实例。我希望获取在过去4小时内插入/更新的所有记录(可配置)，创建一个与每个表相关的csv文件，并将文件存储在S3中。我希望整个过程都是事务性的。如果在从一个表中获取数据时有任何错误，我不希望其他29个表的相关数据被保存在S3中。数据不是很大，每个表中的记录应该在4个小时内保持在100条或更少的顺序。
我正在考虑在EMR集群中设置一个spark作业，从RDS中获取数据，为每个表创建一个csv，并在进程结束时将所有文件发送到S3。一旦数据发布到S3，电子病历集群将被销毁。云监视触发器将每4小时调用一次lamda
浏览 0提问于2018-05-16得票数 0
回答已采纳
3回答
我可以使用spark-csv将表示为字符串的CSV读取到Apache Spark中吗
apache-spark、apache-spark-sql、spark-csv
我知道如何使用spark-csv ()将csv文件读取到spark中，但我已经将csv文件表示为字符串，并希望将此字符串直接转换为数据帧。这个是可能的吗？
浏览 4提问于2016-08-24得票数 7
回答已采纳
1回答
火花:火花-csv花了太长时间
csv、apache-spark、pyspark
我正试图使用Databricks DataFrame包和，从EMR集群上的S3上的CSV源创建一个
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
df = sqlContext.read.format('com.databricks.spark.csv').options(header='true').load('s3n://h2o-airlines-unpacked/allyears.csv')
df.first()
这不会在4m3.xlarge的集群上终止。我
浏览 0提问于2015-08-28得票数 3
2回答
助推spark.yarn.executor.memoryOverhead
amazon-web-services、apache-spark、pyspark、emr、amazon-emr
我试图在EMR上运行一个(py)Spark作业，它将处理大量的数据。当前，我的工作失败了，有以下错误消息：
Reason: Container killed by YARN for exceeding memory limits.
5.5 GB of 5.5 GB physical memory used.
Consider boosting spark.yarn.executor.memoryOverhead.
因此，我谷歌了如何做到这一点，并发现我应该传递带有--conf标志的spark.yarn.executor.memoryOverhead参数。我就是这样做的：
aws emr add
浏览 4提问于2016-06-29得票数 13
回答已采纳
1回答
使用kms加密密钥将数据帧作为csv写入S3，而不提供密钥
scala、apache-spark、amazon-s3、amazon-emr、amazon-kms
我已经通过spark dataframe创建了CSV文件，这些文件会自动加密KMS。
作为参考，我给出了一个创建这些KMS加密文件的示例代码片段。如果你在写的时候看到，我不会给任何KMS密钥。如果你说出根本原因，那将会很有帮助。
val df=spark.read.format("csv").option("header", "true").load("s3:///test/App_IP.csv")
df.createOrReplaceTempView("test")
val df1=spark.sql("
浏览 2提问于2021-01-26得票数 1
2回答
批量数据从关系型数据库转换到S3的最佳方法
amazon-s3、rdbms
我要求将数据(频率=每晚一次传输)从本地关系数据库(SQL server)传输到AWS (作为csv文件和csv格式是必需的)。数据大小约为500 GB。在没有CDC的情况下，每天晚上都会复制整个数据库。
当然，我们可以考虑像parquet或orc这样的格式，但csv在我们的用例中是必需的。
在将数据转换为csv并将其发送到S3之前，我曾考虑将JDBC链接到EMR-spark集群，以便从本地RDBMs数据库导入数据，但我不确定这是否是最优的体系结构。
对于JDBC，我不知道是否有一种方法可以并行使用多个连接(顺便说一句，emr-spark和sqlserver)来加快传输速度。
它是连接本地站点
浏览 4提问于2020-05-03得票数 0
1回答
从AWS向aws EMR集群提交spark作业
amazon-web-services、amazon-s3、pyspark、aws-cli、emr
我正在尝试从命令行aws中正确地向我的aws集群添加一个火花步骤。
一些背景：
我有一个大的数据集(数千个.csv文件)，我需要阅读和分析。我有一个python脚本，看起来类似于：
analysis_script.py
import pandas as pd
from pyspark.sql import SQLContext, DataFrame
from pyspark.sql.types import *
from pyspark import SparkContext
import boto3
#Spark context
sc = SparkContext.getOrCreate(
浏览 2提问于2018-06-04得票数 0
2回答
如何在使用schema Spark读取csv时删除格式错误的行？
scala、csv、apache-spark、apache-spark-dataset
当我使用Spark DataSet加载csv文件时。我更喜欢清晰地指定模式。但是我发现有一些行与我的模式不兼容。列应该是双精度的，但有些行是非数字值。是否可以很容易地从DataSet中过滤出所有不符合我的模式的行？
val schema = StructType(StructField("col", DataTypes.DoubleType) :: Nil)
val ds = spark.read.format("csv").option("delimiter", "\t").schema(schema).load("f.
浏览 4提问于2018-04-09得票数 9
回答已采纳
2回答
S3精选会加速镶木地板文件的火花分析吗？
apache-spark、amazon-s3、parquet
您可以使用S3 Select with Spark on Amazon EMR和with Databricks，但只能用于CSV和JSON文件。我猜S3精选不是为列文件格式提供的，因为它不会有太多帮助。 假设我们有一个包含first_name、last_name和country列的数据湖。 如果数据存储为CSV文件，并且您运行peopleDF.select("first_name").distinct().count()这样的查询，那么S3将把所有列的所有数据传输到ec2集群以运行计算。这真的很低效，因为我们不需要所有的last_name和country数据来运行这个查询。
浏览 16提问于2019-04-30得票数 4
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
