1回答
我正在尝试在每批spark streaming作业中查找数据帧的大小。我能够成功地在批处理作业中找到大小，但当涉及到流时，我无法做到这一点。我一直在数据库上开发spark应用程序，并尝试在流式作业中使用'df.queryExecution.optimizedPlan.stats.sizeInBytes‘。}.start.option(&
浏览 26提问于2019-01-18得票数 0
我正在使用SimpleConsumer，并试图使用spark从消息大小(字节)中提取消息大小。
我能够使用元数据请求获得最早和最新的偏移量，但不知道如何在kafka (0.8.0)中获取字节数。我真的不想使用fetch请求，因为想要检查足够的数据，那么只会运行spark作业(而不是流)来执行操作。
浏览 6提问于2015-07-16得票数 1
回答已采纳
我有一个包含两列的spark数据帧，'keyCol‘列和'valCol’列。数据帧的大小非常巨大，接近1亿行。我想以小批量的方式向kafka主题写入/产生数据帧，即每分钟10000条记录。此spark作业将每天运行一次，从而创建此数据帧 如何在下面的代码中实现每分
浏览 10提问于2020-04-27得票数 1
我有一个Spark结构的流媒体应用程序。应用程序从kafka接收数据，并且应该使用这些值作为参数来处理来自cassandra数据库的数据。我的问题是，我如何使用输入数据帧(kafka)中的数据，作为cassandra中的"where“参数"select”，而不会出现以下错误：
Exception in thread &qu
浏览 25提问于2021-05-24得票数 0
回答已采纳
我是Apache Spark的初学者。我正在尝试运行一个流作业，它接收一些数据，将其转换为数据帧，并运行一些处理，如连接和删除重复项等。现在我必须缓存这个处理过的数据，这样我就可以将它附加到下一个数据流(使用一些联合/连接)，并再次进行处理。
我尝试使用dataframe.cache()缓存并在下一个流批处理中重用它。例如，如果df是从
浏览 0提问于2017-02-16得票数 1
1回答
需要在结构化流中对UDF中的表执行一些查询。问题是，在UDF内部，如果我尝试使用spark.sql，就会得到空指针异常。这里最好的方法是什么。基本上，我需要从一个表中流，然后使用这些数据从另一个表执行一些范围查询。val query = "select count(*) from appleMart where appleId='"+appleId+"'"
val appleCount = s
浏览 3提问于2020-07-29得票数 2
1回答
我是Azure Data Factory的新手。在此之前，我一直在使用SSIS和Pentaho。最近，我开始使用这个工具来创建一些ETL，并且我注意到在过程结束时提供的时间值之间存在一些差异。所以我想知道它们是什么意思(持续时间-处理时间-时间)，特别是为什么持续时间和处理时间之间的巨大差异，这种差异是工具的标准准备时间还是类似的东西？ ? 致以问候。
浏览 22提问于2020-06-17得票数 0
2回答
我正在开发一个SparkStreaming解决方案，在这个解决方案中，一个配置单元表被缓存为一个数据帧，然后流事件将与缓存的数据帧结合起来，以产生结果。然而，问题是Hive表将以特定的频率更新，因此缓存的Dataframe也应该相应地更新。有没有任何方法，比如说TTL或任何其他围绕缓存数据帧的机制，其中数据以特定的时间间隔自动刷新更新的<
浏览 0提问于2018-07-04得票数 0
我有Spark SQL查询，通过几个小的拼图文件(大约2M)。 我有Spark块大小为256M，所以我想将这些小文件组合成一个(或更多)大小为256M的文件。我正在考虑的是找到数据帧大小并将其除以256M，这样我就知道有多少文件将是他们的，但不幸的是Spark不支持查找数据帧大小，因为它是分布
浏览 15提问于2019-06-19得票数 0
回答已采纳
1回答
我是这个领域的新手。我正在尝试将数据从spark推送到phoenix。数据的大小超过100万。spark作业在10万条记录的情况下运行良好，但当记录数超过100万条时就会卡住。我正在从hive加载数据 val hive_data = spark.sql(query) 把它推向凤凰城 hive_data.write.format("org.apache.phoenix.spark</
浏览 26提问于2021-06-26得票数 0
5回答
我正在使用从JSON事件流转换而来的Dataframe来处理事件，这些数据帧最终会被写成Parquet格式。但是，一些JSON事件在键中包含空格，我希望在将其转换为Parquet之前记录和过滤/删除数据帧中的此类事件，因为;{}()\n\t=在拼图方案(CatalystSchemaConverter)中被视为特殊字符，如中所列1下面因此不应允许在列名中使用。如何在Dataframe中对列名
浏览 188提问于2016-07-05得票数 17
2回答
当我运行我的火花作业时，我得到了以下错误：
def aggregateByValue(): DataFrame = {
df.withColumn(&q
浏览 0提问于2018-02-12得票数 4
回答已采纳
我正在调查、Google、Dataflow、和Apache ，以决定哪种解决方案更适合我们的大数据分析业务需求。我发现在spark平台中有Spark SQL和MLlib来进行结构化数据查询和机器学习。
我想知道谷歌数据流平台中是否有相应的解决方案？
浏览 2提问于2015-11-04得票数 32
回答已采纳
在spark批处理作业中，我通常会将JSON数据源写入到一个文件中，并且可以使用DataFrame阅读器的损坏列功能将损坏的数据写出到不同的位置，并使用另一个阅读器从同一作业中写入有效数据。(数据写为拼接) 但在Spark Structred Streaming中，我首先通过kafka将流作为字符串读取，然后使
浏览 34提问于2018-12-27得票数 1
回答已采纳
显然，来自Spark2.4的MLLib的LSHModel支持spark Structured Streaming ()。我在网上找不到更多关于它的信息。有人能帮帮我吗？
浏览 1提问于2021-03-02得票数 0
3回答
使用Smark-SQL2.4.1和kafka进行实时流。我有下面的用例
需要从hdfs加载元数据，以连接kafka的流数据。流式数据记录的特定列应该在元数据数据特定列(Col)中查找。如果发现没有找到元数据列(Col)数据，则将流记录/列数据插入到元数据数据中</
浏览 0提问于2019-08-16得票数 1
回答已采纳
我正在使用spark执行连接倍数表。其中一个表非常大，其他表很小(10-20个记录)。实际上，我想使用包含一对键值的其他表来替换最大表中的值。我的问题是；，这是加入表的最佳方式。(假设有100个或更多的小表) 1)收集小数据，把它转换成地图，广播地图，一步一步地转换大数据。spark.sql(" from bigtable
left join s
浏览 1提问于2018-02-13得票数 3
我正在向yarn上的远程Spark集群提交一个spark作业，并在spark-submit --file中包含一个文件，我希望将提交的文件作为数据帧读取。但我对如何在不将文件放入HDFS中的情况下进行此操作感到困惑：--class com.Employee \--files /User&#
浏览 0提问于2018-11-28得票数 1
1回答
我们有一些这种格式的数据： <event timestamp="20200529032100"> <row attr="gfds" id="6">data</row>
<row attr="sadf" id="8"&
浏览 1提问于2020-05-29得票数 0
我使用的是Cloudera 5.8.3发行版，它是用spark 1.6.0构建的。我通过oozie提交了一个Spark作业(它是作为一个java操作提交的)然而，我也知道spark为<
浏览 0提问于2017-02-07得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
