Spark解析和处理文件parquet/json
Spark是一个开源的分布式计算框架，用于大规模数据处理和分析。它提供了高效的数据处理能力，支持多种数据源和数据格式。在Spark中，parquet和json是常见的文件格式，用于存储和处理结构化数据。
Parquet是一种列式存储格式，它将数据按列存储，提供了高效的压缩和查询性能。Parquet适用于大规模数据集的存储和分析，特别是在需要快速查询特定列或子集的情况下。Parquet文件可以通过Spark读取和解析，然后进行各种数据处理操作。
JSON（JavaScript Object Notation）是一种轻量级的数据交换格式，易于人类阅读和编写，并且易于解析和生成。JSON文件通常用于存储半结构化数据，如日志文件、配置文件等。Spark可以读取和解析JSON文件，并将其转换为DataFrame或RDD进行进一步的数据处理和分析。
对于Spark解析和处理parquet/json文件，可以使用Spark的DataFrame API或RDD API进行操作。DataFrame API提供了更高级的抽象，可以直接读取parquet/json文件并将其转换为DataFrame对象，然后可以使用各种内置函数和操作符进行数据处理和转换。RDD API则提供了更底层的操作接口，可以手动解析和处理parquet/json文件。
在腾讯云中，可以使用腾讯云的云服务器CVM来部署Spark集群，并使用腾讯云对象存储COS来存储和管理parquet/json文件。腾讯云还提供了云原生数据库TDSQL、云数据库CDB等用于存储和管理数据的产品，可以与Spark集成使用。具体的产品介绍和链接地址如下：
- 腾讯云云服务器CVM：提供高性能、可扩展的云服务器实例，用于部署Spark集群。详细信息请参考：腾讯云云服务器CVM
- 腾讯云对象存储COS：提供安全、稳定、低成本的云端存储服务，可用于存储和管理parquet/json文件。详细信息请参考：腾讯云对象存储COS
- 腾讯云云原生数据库TDSQL：提供高性能、高可用的云原生数据库服务，适用于大规模数据存储和查询。详细信息请参考：腾讯云云原生数据库TDSQL
- 腾讯云云数据库CDB：提供稳定可靠的关系型数据库服务，可用于存储和管理结构化数据。详细信息请参考：腾讯云云数据库CDB
通过使用腾讯云的相关产品，结合Spark的强大数据处理能力，可以实现高效、可扩展的parquet/json文件解析和处理。
相关·内容
1回答
我在通过火花阅读地板时遇到了一个问题。
使用类型为a的字段Integer编写了一个拼花文件。然后，使用a的模式作为Long读取这个文件会出现异常。
原因:未实现类型: org.apache.spark.sql.execution.datasources.parquet.VectorizedColumnReader.readIntBatch(VectorizedColumnReader.java:397) at org.apache.spark.sql.execution.datasources.parquet.VectorizedColumnReader.readBatch(Vectoriz
浏览 0提问于2018-08-31得票数 4
回答已采纳
1回答
我想给Parquet写一个这样的Dataframe：
| foo | bar |
|-----|-------------------|
| 1 | {"a": 1, "b": 10} |
| 2 | {"a": 2, "b": 20} |
| 3 | {"a": 3, "b": 30} |
我是用潘达斯和拼花做的：
df = pd.DataFrame({
"foo": [1, 2, 3],
"bar": [{&
浏览 8提问于2020-02-14得票数 4
回答已采纳
我正在尝试读取快速压缩的地板文件，但继续获得以下异常。我找不到这一例外的根本原因，有人能指点我吗？
val sparkSession: SparkSession = SparkSession.builder()
.master("local[2]")
.config("spark.ui.enabled",false)
.appName("local-intellij")
.getOrCreate()
val df = sparkSession.read.parquet("C:\\data\\parquet\\part
浏览 8提问于2020-05-01得票数 1
回答已采纳
1回答
我在亚马逊s3上使用pyspark读取s3存储桶上的对象。我的存储桶是由许多json文件组成的，我读取了这些文件，然后将其另存为拼图文件
spark.read.json('s3://my-bucket/directory1/')
spark.write.parquet('s3://bucket-with-parquet/', mode='append')
每天我都会上传一些新的文件到s3://my-bucket/directory1/上，我想更新它们到s3://bucket-with-parquet/，有没有一种方法可以确保我不会两次更新数据。我
浏览 1提问于2019-12-03得票数 2
1回答
我正试着读一本“星星之火”，并把它写回拼花。我正在windows中运行我的代码。下面是我的密码。执行之后，它创建一个名为output_spark.parquet的文件夹。它还引发一个错误，即找不到该文件。如果我创建了一个文件，然后运行代码，它会说该文件已经存在。这是我所犯的错误。
py4j.protocol.Py4JJavaError:调用o34.parquet时出错。：java.lang.RuntimeException: java.io.FileNotFoundException: java.io.FileNotFoundException: HADOOP_HOME和hadoop.hom
浏览 2提问于2021-03-03得票数 0
回答已采纳
我想将Elasticsearch索引存储到HDFS文件中，而不是使用ES-Hadoop连接器。建议的解决方案是使用流式定制接收器读取并保存为拼图文件，并且代码如下所示， JavaDStream<String> jsonDocs = ssc.union(dsList.get(0), dsList.subList(1, dsList.size())); // I have a couple receivers
jsonDocs.foreachRDD( rdd -> {
Dataset<Row> ds = spark.read().json(spark.cre
浏览 28提问于2019-04-02得票数 1
1回答
拼图文件被读取为空
、、、
我正在尝试读取我从我的Jupyter笔记本上的HDFS下载的parquet文件，但是它显示为空。我知道它不是空的，因为我在将它保存到HDFS之前已经处理过它。有人知道为什么它被读为空吗？
HDFS和群集环境中文件的大小：
hadoop fs -du -s -h /user/some/test.parquet
1.2 M 3.5 M /user/some/test.parquet
val test = spark.read.parquet("hdfs:///user/some/test.parquet")
test.count()
res0: Long = 10
在Jup
浏览 0提问于2021-06-18得票数 0
我想把二进制格式的拼花文件转换成csv文件。我正在使用以下命令的火花。
sqlContext.setConf("spark.sql.parquet.binaryAsString","true")
val source = sqlContext.read.parquet("path to parquet file")
source.coalesce(1).write.format("com.databricks.spark.csv").option("header","true").save(&
浏览 0提问于2018-10-01得票数 3
回答已采纳
当我尝试使用pyspark从本地目录保存拼图文件时，出现以下错误。我尝试了spark 1.6和2.2，两者都给出了相同的错误
它正确地显示了模式，但在写入文件时出现错误。
base_path = "file:/Users/xyz/Documents/Temp/parquet"
reg_path = "file:/Users/xyz/Documents/Temp/parquet/ds_id=48"
df = sqlContext.read.option( "basePath",base_path).parquet(reg_p
浏览 0提问于2017-11-10得票数 0
文件夹结构：
A -> B1->C1->.json
-> B2->C2->.json
A和B下可以有许多文件夹，它不遵循任何模式。
以上是ADLS中的文件夹结构，当使用spark递归读取Json文件时，我们得到了下面的错误。
java.net.URISyntaxException :绝对路径URI中的相对路径
def json_parquet(sourceFilePath):
df=(spark.read.format("json")
.option("multiline",True)
.
浏览 5提问于2022-12-03得票数 0
我正在尝试使用Spark将一堆csv文件转换为parquet，有趣的是，输入的csv文件已经按目录“分区”了。所有输入文件都有相同的一组列。输入文件结构如下：
/path/dir1/file1.csv
/path/dir1/file2.csv
/path/dir2/file3.csv
/path/dir3/file4.csv
/path/dir3/file5.csv
/path/dir3/file6.csv
我想用Spark读取这些文件，并将它们的数据写入hdfs中的parquet表，保留分区(按输入目录分区)，例如每个分区有一个输出文件。输出文件结构应如下所示：
hdfs://path/di
浏览 2提问于2016-11-15得票数 4
因此，我有一个问题，在编写分区Parquet文件时，DataFrame中的某些行会被删除。
以下是我的步骤：
用指定的模式从S3读取CSV数据文件
按“日期”列(DateType)分区
用mode=append编写Parquet
阅读的第一步如预期，没有解析问题。为了进行质量检查，我执行以下操作：
对于date='2012-11-22'的特定分区，对CSV文件、加载的DataFrame和parquet文件执行计数。
下面是一些代码，可以使用pyspark来再现：
logs_df = spark.read.csv('s3://../logs_2012/
浏览 0提问于2017-11-02得票数 2
回答已采纳
我的团队正在构建一个ETL流程，使用Spark将原始分隔文本文件加载到基于Parquet的“数据湖”中。Parquet列存储的承诺之一是，查询将只读取必要的“列条纹”。
但是，我们看到嵌套架构结构正在读取意外的列。
为了演示，这里是一个使用Scala和Spark 2.0.1 shell的POC：
// Preliminary setup
sc.setLogLevel("INFO")
import org.apache.spark.sql.types._
import org.apache.spark.sql._
// Create a schema with nested c
浏览 31提问于2016-10-22得票数 22
这里是Spark的新用户。我通过Spark找不到任何关于同一个dataFrame的JSON和拼图输出的文件大小比较的信息。
现在用一个非常小的数据集进行测试，执行一个df.toJSON().collect()，然后写入磁盘创建一个15kb的文件。但是执行df.write.parquet会创建105个文件，每个文件的大小约为1.1kb。为什么在这种情况下，使用parquet的总文件大小要比使用JSON大得多？
提前感谢
浏览 20提问于2018-08-14得票数 2
我有一个数据帧，当保存为Parquet格式时，使用的是~11 as。当阅读数据文件并写入json时，需要5分钟。当我添加partitionBy("day")时，它需要几个小时才能完成。我知道分配到分区是代价高昂的操作。有办法让它更快吗？对文件进行排序会使它变得更好吗？
示例：
运行5分钟
df=spark.read.parquet(source_path).
df.write.json(output_path)
运行了几个小时
spark.read.parquet(source_path).createOrReplaceTempView("source_table
浏览 3提问于2017-07-23得票数 8
回答已采纳
我有一个简单的pyspark程序，一次读取2个文本文件，将每一行转换为json对象，并将其写入拼图文件，如下所示：
for f in chunk(files, 2):
file_rdd = sc.textFile(f)
df = (file_rdd
.map(decode_to_json).filter(None)
.toDF(schema)
.coalesce(5)
.write
.partitionBy("created_year
浏览 0提问于2016-12-10得票数 0
我想用scala编写简单的JSON字符串到parquet和avro文件格式，而不使用spark框架。
我的JSON字符串看起来：
{"emp_id":"123","emp_name":"Mike","emp_status":"true"}
我没有找到任何解决方案，有没有可能在scala中使用简单的JSON字符串编写parquet和avro文件，而不需要星火框架？
浏览 6提问于2022-02-08得票数 0
回答已采纳
我必须使用json文件中的信息创建一个定制的org.apache.spark.sql.types.StructType模式对象，json文件可以是任何内容，所以我已经在属性文件中对它进行了参数化。
以下是属性文件的外观：
//ruta al esquema del fichero output (por defecto se infiere el esquema del Parquet destino). Si existe, el esquema será en formato JSON, aplicable a DataFrame (ver StructType.fromJson)
sch
浏览 6提问于2016-11-10得票数 3
回答已采纳
我正在读取镶木地板文件，并将其转换为JSON格式，然后发送给kafka。问题是，它读取了整个拼图，所以一次性发送给kafka，但我想逐行或分批发送json数据： object WriteParquet2Kafka {
def main(args: Array[String]): Unit = {
val spark: SparkSession = SparkSession
.builder
.master("yarn")
.appName("Write Parquet to Kafka")
.getO
浏览 18提问于2019-01-22得票数 1
回答已采纳
我知道Sqoop支持将数据导入为Avro、Parquet、Text等格式，有没有办法将数据导入为JSON？
使用Spark目前对我来说不是一个选择。
浏览 0提问于2019-02-27得票数 0
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回答
阅读并加入几个拼花文件
、、、
我有几个拼板文件，我想读和加入(合并在一个文件中)，但我使用的是一个基本的解决方案，我认为这不是最好的。
每个文件都有两个用于连接的id变量和一个变量，它在每个拼板中都有不同的名称，因此要将所有这些变量都放在同一块中。
多个表(n表variables)final n1表(id +n变量)
我正在使用的解决方案
file_path_list = ["file1.parquet","file2.parquet","file3.parquet"]
# read first file
df = spark.read.format('p
浏览 7提问于2022-06-02得票数 0
回答已采纳
1回答
我有带有空数组(array (nullable = true)和element: (containsNull = true))的JSON文件，我想将它们转换为parquet文件。这些空字段将自动删除，而所有其他列都会按预期进行转换。是否有一种方法可以用其他方法替换空数组(例如["-"])？我在AWS中运行我的代码，但是替换将使用纯PySpark和dataframes进行。
数据类型仍然必须是parquet文件中的数组(只是不带空值)。
杰森：
{
"a":[],
"b":[1,2],
"c": "a string&#
浏览 3提问于2022-01-19得票数 1
回答已采纳
在Databricks上，我有一个每天运行的作业，它将数据帧写入Azure Blob存储上的一个拼花文件。它创建一个文件和一个名称相同的文件夹。但是，在我将其设置为空的目录上创建的文件是空的，并且有一个名称相同的文件夹，其中包含一个带有“随机名称”的parquet文件，该文件存储dataframe的内容。
我希望将拼花文件保存为agg_DF.parquet，以便文件名保持不变，但agg_DF.parquet现在是空的。相反，看起来我需要进入文件夹并获取以.parquet 结尾的任何文件，但是我真的很想知道如何做这个。或者，是否有更好的方法在Databricks上这样做，以便agg_
浏览 12提问于2022-06-30得票数 0
从读取数据流到Apache的标准方法是：
events = spark.readStream \
.format("json") \ # or parquet, kafka, orc...
.option() \ # format specific options
.schema(my_schema) \ # required
.load("path/to/data")
但是在应用架构之前，我需要清理一些数据重新排列一些字段，我希望会有一个
events = spark.readS
浏览 4提问于2017-11-10得票数 1
回答已采纳
我有一个类似于json的结构，看起来如下：
>>> df = spark.read.parquet(good_partition_path)
id: string
some-array: array
element: struct
array-field-1: string
array-field-2: string
根据分区的不同，some-array可能是所有id的空数组。
>>> df = spark.read.parquet(bad_partition_path)
id: string
some-array:
浏览 8提问于2020-02-19得票数 5
回答已采纳
1回答
我必须读取一个位于不同年份的多个文件夹中的‘’.parket‘’文件。当是一年或两年的时候，这不是问题，但是当超过两年的时候，事情就变得复杂了，因为我必须阅读每个月对应的12个子目录。我举了一个例子，说明我是如何以一种低效的方式做到这一点的。
步骤1:读取文件
2019年
df_2019_01=spark.read.parquet('/2019/01/name.parquet/')
df_2019_02=spark.read.parquet('/2019/02/name.parquet/')
df_2019_03=spark.read.parquet('
浏览 2提问于2021-06-29得票数 0
1回答
我使用Apache 1.9.2将关系数据库中的数据加载到中。其目的是将结果写入Parquet文件中，因为它以柱状方式存储数据。为了实现这一点，我使用了Nifi中的ConvertAvroToParquet (默认设置)处理器(后面是PutGCSObject处理器)。这些结果文件的问题是，当我使用Spark2.4.0(Scala2.11.12)中的文件时，无法读取Decimal类型的列:不能转换.列: ARHG3A，预期:十进制(2，0)，找到:二进制
到parquet/avro示例文件的链接：
正如我所知道的，Nifi在流程文件中与处理器之间的Avro格式一起工作，我还编写了avro文件(就像
浏览 0提问于2019-06-17得票数 3
我们使用的是pyspark 1.6。并试图通过压缩将文本转换为其他文件格式(如Json、csv等) (gzip、lz4、snappy等)。但无法看到压缩的工作原理。
请找到我们尝试过的代码攻击。请帮助我们在我们的代码中指出问题，否则建议解决方法。为了补充这个问题，没有一个压缩在1.6中工作，但它在spark 2.X中工作良好
选项1：
from pyspark import SparkContext SparkConf
sqlContext.setConf("spark.sql.parquet.compression.codec", "snappy")
df
浏览 2提问于2017-07-21得票数 0
因此，我有一个火花结构化流作业，它将扁平的json消息数据存储在hdfs位置中的日期分区文件夹中，即/source/wireless/active_portal/activation/dt=current date，它还在创建_spark_metadata，其中包含它写入输出位置的文件的名称。这个HDFS位置有一个蜂窝表在上面。
我必须在不更改名称的情况下将所有拼板文件中的特定列值PIN更改为默认值XXXX，这样_spark_metadata就不会受到干扰。
是否有一种方法可以通过使用spark命令或使用hive命令在hdfs位置的列中更改数据。
schema of parquet files
浏览 5提问于2022-03-31得票数 1
1回答
通过运行我的scala-spark应用程序，我得到了以下异常： 19/02/22 11:27:48 WARN TaskSetManager: Lost task 0.0 in stage 3.0 (TID 3, localhost): java.lang.OutOfMemoryError: Java heap space
at parquet.column.values.dictionary.IntList.initSlab(IntList.java:90)
at parquet.column.values.dictionary.IntList.<init>(IntList.ja
浏览 30提问于2019-02-23得票数 0
背景：，我编写了一个简单的火花结构化蒸汽应用程序，可以将数据从卡夫卡转移到S3。发现为了准确地支持-一旦保证火花创建_spark_metadata文件夹，最终变得太大，当流式应用程序运行了很长时间，元数据文件夹变得如此大，我们开始得到OOM错误。我想摆脱星火结构化流的元数据和检查点文件夹，并自己管理偏移。
我们如何管理火花流中的偏移：，我已经使用val offsetRanges = rdd.asInstanceOfHasOffsetRanges.offsetRanges来获得Spark结构化流中的偏移量。但想知道如何获得偏移和其他元数据来管理检查点，我们自己使用Spark结构化流。您有实现检查
浏览 2提问于2020-06-17得票数 1
回答已采纳
似乎在Spark 2.1.0中，保存空的Parquet文件被破坏了，因为无法再次读取它们(由于错误的模式推断)。
我发现，从Spark2.0开始，在编写拼图文件时，默认情况下禁用_metadata文件的写入。但是我找不到配置设置来重新打开它。
我尝试了以下几种方法：
spark_session = SparkSession.builder \
.master(url) \
.appName(name) \
.config('spark.hado
浏览 52提问于2017-01-25得票数 3
回答已采纳
3回答
我遇到了一个问题，我在S3中将拼图数据作为每日数据块(以s3://bucketName/prefix/YYYY/MM/DD/的形式)，但我无法从不同的日期读取AWS EMR Spark中的数据，因为一些列类型不匹配，并且我得到了许多异常之一，例如：
java.lang.ClassCastException: optional binary element (UTF8) is not a group
当某些文件中的数组类型具有值，但同一列在其他文件中可能具有null值时出现，这些值随后被推断为字符串类型。
或
org.apache.spark.SparkException: Job aborte
浏览 0提问于2016-12-02得票数 24
回答已采纳
1回答
我有一个json配置文件，我把我的路径存储在那里。
数据是按月和天排列的，因此如果没有json，我将使用f-字符串，如下所示：
spark.read.parquet(f"home/data/month={MONTH}/day={DAY}")
现在我想从json那里提取出来。然而，我遇到了月和日变量的问题。我不想在json中分裂这条路。但是这样写的话：
{
"path":"home/data/month={MONTH}/day={DAY}"
}
并装载：
DAY="1"
MONTH="12"
conf_pat
浏览 2提问于2022-10-18得票数 2
回答已采纳
2回答
我是作为JSON从kafka_2.11-0.10.0.1到Spark2.1的流式计量器读取记录。我转向了结构化流媒体；尽管卡夫卡用户确认了输入的数据，但我的控制台和writeStream没有移动。我正在测试使用
pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0
我的代码：
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
spark = SparkSe
浏览 0提问于2017-12-15得票数 3
回答已采纳
(火花2.0.2)
这里的问题是，当您有不同模式的分割文件并在读取期间强制使用模式时，问题就会出现。即使可以打印架构并运行show() ok，也不能对缺少的列应用任何筛选逻辑。
下面是两个示意图示例：
// assuming you are running this code in a spark REPL
import spark.implicits._
case class Foo(i: Int)
case class Bar(i: Int, j: Int)
因此，Bar包含了Foo的所有字段，并添加了一个(j)。在现实生活中，当您从模式Foo开始，然后决定需要更多的字段并以模式Bar
浏览 8提问于2017-09-08得票数 2
回答已采纳
我遇到了一个问题，我的电火花工作间歇性地失败了。当我使用下面的代码从目录读取所有的"C:\Python27\Lib\site-packages\pyspark\python\lib\pyspark.zip\pyspark\serializers.py"，文件时，它会抛出异常文件EOFError EOFError中的第717行。
inputPathWithData = []
if (path.exists(input)):
inputPathWithData.append(input + "\\*.parquet")
if (le
浏览 2提问于2020-07-02得票数 0
我有一个很大的嵌套NDJ (新行分隔的JSON)文件，我需要将其读取到单个spark dataframe中并保存到parquet中。在尝试呈现模式时，我使用以下函数：
def flattenSchema(schema: StructType, prefix: String = null) : Array[Column] = {
schema.fields.flatMap(f => {
val colName = if (prefix == null) f.name else (prefix + "." + f.name)
浏览 2提问于2016-12-10得票数 5
当我试图读取一个文件夹，这个文件夹应该包含Parquet格式的文件，如果有数据，如果没有数据，那么第一行就会出错，后续代码也不会执行。
val hdfsData: DataFrame = spark.sqlContext.read.parquet(hdfsPath)
hdfsData.rdd.isEmpty() match ....
....
错误: org.apache.spark.sql.AnalysisException:无法推断Parquet的架构。必须手动指定。
处理这种情况的正确方法是什么。
浏览 0提问于2018-11-22得票数 3
回答已采纳
2回答
我很难找到这个问题的答案。假设我向parquet编写了一个数据帧，并结合使用repartition和partitionBy来获得一个分区良好的parquet文件。如下所示：
df.repartition(col("DATE")).write.partitionBy("DATE").parquet("/path/to/parquet/file")
现在，稍后我想要读取拼图文件，所以我这样做：
val df = spark.read.parquet("/path/to/parquet/file")
数据帧是否按"DATE
浏览 5提问于2018-06-13得票数 15
我的数据流是：json -> parquet -> Athena，但是嵌套的键值字段tags有一个问题。
JSON文件是
[{"myid":1,"name":"foo","tags":{"tag1":"a","tag2":"b"}},
{"myid":2,"name":"bar","tags":{"tag1":"c","tag2":"d
浏览 20提问于2019-11-23得票数 0
回答已采纳
2回答
我对scala很陌生，我试图在集群上使用Scala version 2.11.12、spark version 2.4.3和JDK 1.8.0_212运行一个Spark。我的build.sbt是这样的-
name := "LearningScala"
version := "1.0"
scalaVersion := "2.11.12"
val sparkVersion = "2.4.3"
//resolvers += "Maven Dependencies" at "https://repo1.maven
浏览 0提问于2019-07-31得票数 0
回答已采纳
1回答
我正在评估不同方式在星火中加载Parquet文件的性能，差异是惊人的。
在我们的Parquet文件中，我们有以下类型的嵌套案例类：
case class C(/* a dozen of attributes*/)
case class B(/* a dozen of attributes*/, cs: Seq[C])
case class A(/* a dozen of attributes*/, bs: Seq[B])
从Parquet文件加载它们需要一段时间。因此，我对从Parquet文件加载case类的不同方法进行了基准测试，并使用Spark1.6和2.0对字段进行了求和。
以下是我所做
浏览 2提问于2016-08-24得票数 7
我有一个bitnami spark docker基础设施(一个主机和一个工人)。 Spark读取Kafka流。 stream_df = spark.readStream.format("kafka")\
.option("kafka.bootstrap.servers", "kafka1:19091")\
.option("subscribe", "Aleca")\
.option("startingOffsets", "earliest")\
.l
浏览 36提问于2021-04-29得票数 0
我有一个包含多个Int8和字符串列的模式，我已经将它们写入到Parquet格式中，并存储在一个S3A存储桶中供以后使用。
当我尝试使用SqlContext.read.option("mergeSchema","false").parquet("s3a://....")读取这个拼图文件时，我得到了以下异常。
我尝试使用parquet -tools(带有schema和meta选项)来读取parquet文件，但是我得到了一个未知的命令错误。
*Exception in thread "main" org.apache.spark.sql
浏览 1提问于2016-10-04得票数 1
1回答
我正在运行Spark 2，并试图混洗大约5TB的json。在Dataset的混洗过程中，我遇到了很长的垃圾收集暂停
val operations = spark.read.json(inPath).as[MyClass]
operations.repartition(partitions, operations("id")).write.parquet("s3a://foo")
有没有什么明显的配置调整来处理这个问题？我的配置如下：
spark.driver.maxResultSize 6G
spark.driver.memory 10G
spark.execu
浏览 12提问于2016-08-17得票数 5
回答已采纳
我是Apache Spark 1.3.1的新手。如何将JSON文件转换为Parquet？
浏览 2提问于2016-01-12得票数 10
回答已采纳
1回答
我有数以百万计的流媒体拼花文件正在编写中。为了调试和分析目的，我想支持运行临时交互查询(如果我也可以运行流查询来实时监视关键指标)。什么是可扩展的解决方案来支持这一点。我观察到的两种方法是在数百万个parquet文件上交互地运行spark (不太熟悉spark生态系统，但这是否意味着为每个sql用户提交的文件运行spark作业，还是需要运行一些流作业并以某种方式提交查询)和第二种方式是在parquet之上使用presto sql引擎(不确定预置如何获取新传入的parquet文件)。任何一种方法的任何建议或利弊。考虑到我每天都有>~10 day的数据，任何更好的解决方案。
浏览 0提问于2018-01-19得票数 0
1回答
是否有方法为星火结构化文件流源指定起始偏移量？
我正在尝试从HDFS源源不断地发送文件：
spark.sql("SET spark.sql.streaming.schemaInference=true")
spark.readStream
.parquet("/tmp/streaming/")
.writeStream
.option("checkpointLocation", "/tmp/streaming-test/checkpoint")
.format("parquet")
.opti
浏览 1提问于2018-07-18得票数 3
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储活动推荐
腾讯云开发者
