在批处理之间持久存在的内存缓存中触发结构化流
，是指在批处理任务之间通过持久性内存缓存触发结构化流式处理的一种方式。结构化流处理是一种用于处理连续数据流的计算模型，能够在数据到达时进行实时处理和分析。
通过在批处理任务之间使用持久性内存缓存，可以将数据暂存于内存中，以提高数据的访问速度和处理效率。这种方式可以减少对磁盘和网络的访问，从而加快数据的处理和分析过程。
结构化流处理可以应用于许多场景，如实时分析、实时报表生成、实时推荐系统、实时监控等。它适用于对实时数据进行持续处理和分析的场景，能够快速响应数据的变化，并实时生成结果。
腾讯云提供了一些相关的产品和服务来支持在内存缓存中触发结构化流处理，如腾讯云的数据处理服务、流计算服务等。这些产品和服务能够帮助用户快速构建和部署结构化流处理应用，提供高性能、可靠性和可扩展性的计算能力。
以下是一些腾讯云的相关产品和产品介绍链接地址：
- 腾讯云数据处理服务：提供了一种灵活、可扩展的数据处理平台，支持批处理和流处理任务的处理和分析。详情请参考：腾讯云数据处理服务
- 腾讯云流计算服务：基于Flink流处理引擎，为用户提供了一种高性能、可扩展的实时数据处理和分析服务。详情请参考：腾讯云流计算服务
通过使用这些腾讯云的产品和服务，用户可以快速构建和部署在内存缓存中触发结构化流处理的应用，实现对实时数据的快速处理和分析。
相关·内容
1回答
我是星火公司的新手，我想了解是否会有额外的开销/延迟来持久化和不持久化内存中的数据。
据我所知，在我们使用缓存数据saved时没有发生数据移动，它只是保存在executor的内存中。因此，这应该只是一个设置/取消一个标志的问题。
我在火花流作业中缓存数据帧，并想知道这是否会导致批处理执行中的额外延迟。
浏览 2提问于2020-04-07得票数 1
回答已采纳
我有一个Spark集群，总共有17个executors。我已经将Spark 2.1与Kafka集成在一起，并从topic中读取数据，例如：
val df = spark
.readStream
.format("kafka")
.options("kafka.bootstrap.servers","localhost:9092")
.options("subscribe","test")
.load
现在我想知道，当我以集群模式提交我的spark应用程序时，将分配多少个执行者(总共17个)来监
浏览 0提问于2018-09-05得票数 2
我遇到了一些东西，乍一看，星火流新手似乎是违反直觉的：
当火花结构化流开始处理更多的数据时，它的批处理持续时间减少了
这可能不是最精确的图片，但我看到了更清晰的图案。
我可能需要解释一下，批处理持续时间到底是什么?我的理解是，它代表了火花处理流的迷你批处理所需的秒数。
接下来，我需要澄清是如何触发小型批处理的--无论它是基于批处理中的数据量还是时间间隔.
编辑
代码如下。有相当多的“重”操作(联接、dropDuplicates、使用HOF、udfs、.)。汇和源都是蓝色的晚霞
# [CONFIGS]
ehConfig = {
'eventhubs.startingPosition
浏览 1提问于2019-09-23得票数 0
1回答
我正在创建一个应用程序，从服务器下载六张最新图片(缩略图)，然后通过Viewpager下载。回放时显示了前六位。储存它们的最佳做法是什么？我应该将它们保存在sqlite数据库中并在会话后删除，还是应该将它们存储在堆中？目前，我将它们存储在堆中，当下载42个缩略图时，它会销毁前六个。
浏览 5提问于2013-01-06得票数 0
我有一个结构化的流媒体应用程序读取来自Kafka的消息。每天的信息总数约为180亿条，每分钟的峰值信息计数为12,500,000条。最大消息大小为2KB。
如何确保我的结构化流应用程序能够处理如此大的数据量和速度？基本上，我只想知道如何设置最佳触发时间、maxOffsetsPerTrigger或任何其他配置，从而使作业顺利进行，并能够处理故障和重新启动。
浏览 2提问于2019-06-17得票数 5
我正在努力使流式和批处理数据文件的联合操作。但是，我发现了以下错误：
无法从用户代码中抛出错误: org.apache.spark.sql.AnalysisException:不支持流和批处理DataFrames/数据集之间的联合；
这个特性在Stream中还不支持吗？
谢谢。
浏览 0提问于2019-08-23得票数 1
回答已采纳
我们有一个正在进行的批处理，如下所述。
Hive SQL用于每日批处理。
数据被从文件或RDMBS中摄取
数据是在Raw --> Staging -> Mart中摄取的，从暂存到mart是所有的业务转换，而对暂存的原始处理只是数据的清理和格式化。
现在，作为获取真实或接近实时数据的一部分，我正在评估Lambda体系结构，这是什么计划？
所有的源头系统都会落在卡夫卡身上。
同一批处理系统将消耗卡夫卡主题。
新的火花应用程序将消耗卡夫卡主题的流。
服务层将创建视图，这些视图将流和批处理的聚合数据结合在一起进行真实(接近真实的)时间处理。
的问
浏览 2提问于2018-09-18得票数 1
1回答
流处理和批处理
、、、、
由于批处理的延迟是在累积特定数目的数据时产生的，我是否可以将“一的大小”的批处理视为流处理？或者当运算符进行计算时还有其他的区别吗？
例如，如果我将基于火花的程序的批处理大小设置为1，是否可以使其延迟降低到flink？
我的一种想法是:对于流处理，如果处理了前一个操作符，一个数据流就会从前一个操作符流到后一个操作符，但是对于批处理过程，只有在操作符处理完一个数据之后，它才能接受另一个数据。流处理中的流水线似乎是加速的关键。
我的解释正确吗？如果错了，对我的问题有什么适当的解释。
浏览 9提问于2022-11-14得票数 0
集群中有RabbitMQ服务器(2个节点)。所有队列都是持久的，镜像的，所有消息都被设置为持久的。
我编写了应用程序来同步RabbitMQ队列上的数据库更改。
在大多数情况下，队列是空的，因为使用者可以像生产者一样快速地读取更改。
不幸的是，在初始同步(当所有表中的所有行都被传输时)，队列中有大量消息(例如，10 be )等待使用，因为从数据库读取数据在大多数情况下比写入快。我以为这些消息将被保存到磁盘中，但似乎所有的消息都存储在RAM中。因此，当所有RAM都被使用时(不管我有多少内存)，它就开始阻止发行商了。
有谁知道为什么RabbitMQ也将所有持久和持久的消息保存在内存中？这是“按设计”
浏览 0提问于2014-02-13得票数 2
我有一个Azure数据湖存储容器，它充当由Apache处理JSON文件的着陆区。
那里有数以万计的小文件(最多几MB)。火花代码定期读取这些文件，并执行一些转换。
我希望文件被正确读取一次，而火花脚本是幂等的。如何确保文件不被一次又一次地读取？我怎样才能有效地做到这一点呢？
我是这样看数据的：
spark.read.json("/mnt/input_location/*.json")
我考虑了以下几种方法：
创建一个带有已处理的文件名的Delta表，并在输入DataFrame上运行EXCEPT转换
将处理的文件移动到不同的位置(或重命名它们)。我不想那样做。如果我需要重
浏览 2提问于2021-04-28得票数 3
回答已采纳
我正在试着找出什么是更好的方法。
我有一个火花批处理作业，这是计划运行每5分钟，它需要2-3分钟来执行。
由于已经添加了对动态分配spark.streaming.dynamicAllocation.enabled的支持，将其设置为每5分钟从源拉取数据的流式作业是一个好主意吗？
在流/批处理作业之间进行选择时，我应该记住哪些事项？
浏览 0提问于2019-07-24得票数 0
我正在研究将Apache Spark用于一个应用程序。我对使用临时视图和完整SQL查询的结构化流模式特别感兴趣(为了简单和低延迟)。 应用程序将需要在单个输入数据流上运行多个(数十个，可能数百个)查询。有没有办法避免Spark重复读取每个查询的输入？
浏览 27提问于2021-01-20得票数 0
回答已采纳
当使用核心数据时，我发现了一个问题，我用上下文保存保存了X个对象，一切正常，直到我关闭应用程序并再次打开它，于是所有相同的对象都有" data : works“。我已经尝试在applicationWillResignActive上保存上下文，只是为了确保堆栈已经保存，但是没有用。当我在应用程序退出活动期间检查对象时，它们是好的，但在应用程序变为活动状态时检查它们会显示故障。我漏掉了什么吗？我不确定这些核心数据对象是如何在应用程序退出时出错的。
浏览 0提问于2014-05-29得票数 0
1回答
我开始学习火花，并有一个困难的时间理解背后的合理性结构化流在星火。结构化流将到达的所有数据视为无界输入表，其中数据流中的每个新项都被视为表中的新行。下面的代码将在传入的文件中读取到csvFolder。
val spark = SparkSession.builder.appName("SimpleApp").getOrCreate()
val csvSchema = new StructType().add("street", "string").add("city", "string")
.add("z
浏览 2提问于2017-05-20得票数 5
回答已采纳
1回答
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Runtime.Caching;
namespace Caching_IApiOutputCache_MemeoryCacheDefault
浏览 1提问于2014-04-06得票数 0
1回答
我正在尝试将数据写入一个卡夫卡主题后，阅读了一个蜂巢表如下。
write_kafka_data.py:
read_df = spark.sql("select * from db.table where some_column in ('ASIA', 'Europe')")
final_df = read_df.select(F.to_json(F.struct(F.col("*"))).alias("value"))
final_df.write.format("kafka")\
浏览 0提问于2021-09-13得票数 4
回答已采纳
1回答
我有类可以从db和servlet获取数据，以便将这些数据发送到jsp。如果我在表中插入或删除行(使用pgAdmin)，jsp上的数据将被更新(使用新数据)，但是如果我在表中更新现有日期，则jsp上不会更新它(仅在重新启动glassfish之后)。类，用于ORM：
package db_classes;
@Entity
public class heading {
private Integer id;
private String name;
private Long themeCount;
private Collection<topic> topicsById;
@Id
@C
浏览 0提问于2015-02-25得票数 0
回答已采纳
1回答
我将batch_size设置为application.yml：
spring:
profiles: dev
jpa:
properties:
hibernate:
generate_statistics: true
jdbc:
batch_size: 20
为什么batch_size不应该设置为尽可能高的值？将batch_size设置为10000的缺点是什么？对于batch_size: 20 20查询，它的行为不与<= 20相同，对于>20的查询，它总是在一批中发送它们，而不是对数据库执行另一次请求。这
浏览 1提问于2019-12-27得票数 2
回答已采纳
我对结构化流中的startingOffsets感到困惑。在正式文档中，它表示查询类型。
流-这是连续流吗？
批处理-这是用于查询forEachBatch还是触发器？(不允许最近一次)
我的工作流程也启用了checkpoints。如何与startingOffsets一起工作呢？如果我的工作流程崩溃，并且我有startingOffsets作为latest，那么火花检查卡夫卡偏移量还是火花检查点偏移量，还是两者都有？
浏览 4提问于2021-10-03得票数 1
回答已采纳
1回答
我有一个长期运行的结构化流媒体作业，它消耗了几个Kafka主题，并在滑动窗口上聚合。我需要了解如何在HDFS中管理/清理检查点。 作业运行良好，我能够从失败的步骤中恢复，而不会丢失数据，但是，我可以看到HDFS利用率每天都在增加。我找不到任何关于Spark如何管理/清理检查点的文档。以前，检查点存储在s3上，但由于要读/写大量的小文件，这被证明是非常昂贵的。 query = formatted_stream.writeStream \
.format("kafka") \
.out
浏览 34提问于2019-01-07得票数 2
回答已采纳
1回答
我有一个芹菜配置良好，并与django合作。在post_save信号中，我使用一个任务和另一个周期性任务向一个集合发送一个新的记录，我试图使用这个集合。
from __future__ import absolute_import, unicode_literals
from celery import shared_task
class Data():
def __init__(self):
self.slotshandler = set()
global data
data = Data()
@shared_task
def ProcessMailSendin
浏览 0提问于2018-09-15得票数 0
回答已采纳
1回答
持久化Drools知识会话
、、、、
在Drools中持久化知识会话是可能的吗？如果是这样的话，是怎么做的？我将事实存储在数据库中，我希望避免每次添加新事实时都在新会话中重新加载所有事实。此时，当有新的事实时，该事实被存储在数据库中，然后调用Drools :它加载会话中的所有事实，然后触发规则。会话在两次调用之间丢失。
浏览 1提问于2011-06-01得票数 6
1回答
我使用Spark流来处理在线需求，例如每小时新的用户计数如下：
根据每批处理，当日志出现时，从外部表(如hbase或dynamodb )中选择uid，如果不存在，则插入表
这种方法经常使用表格，花费太大。
现在我想用结构化流来解决这个问题。
在下面的sql中，可以脱机地解决问题：
sql1
create table event_min_table as select pageid,uid,floor(min(time)/36000)*3600 as event_time from event_table group by pageid,uid
sql2
select pageid,count(
浏览 3提问于2017-09-08得票数 0
回答已采纳
我从最近的通量开始，需要一些关于如何更新组件状态和存储的建议。
我需要设置一个连接到onclick事件的切换标志，根据我在Flux体系结构中的理解，我应该调用一个操作，然后触发一个分派事件，然后商店将更新并发出更新，从而生成我的组件来接收更新和重新呈现。(如果我错了，请更正)组件代码如下所示：
...
_updateState:function(){
this.setState({myTrigger: MyStore.myTrigger});
},
_onClickEvent: function(boolValue){
MyActions.updateTrigger(boolV
浏览 4提问于2015-03-05得票数 0
回答已采纳
我使用的是Cuda：cudaMemcpyAsync ( void* dst, const void* src, size_t count, cudaMemcpyKind kind, cudaStream_t stream = 0 )
从GPU内存从CPU内存复制数据。如果使用memcpy()将数据从CPU内存复制到持久内存，则需要显式调用刷新操作(例如。确保数据从CPU缓存中刷新。在使用cudaMemcpyAsync()将GPU内存复制到持久内存时，是否需要调用刷新操作；
浏览 28提问于2022-01-08得票数 2
回答已采纳
PHP静态变量(在正常页面请求期间确实持久化)是否在逐步批处理API之间持久化？那么在非渐进(命令行)批处理执行过程中呢？&drupal_static以任何方式发挥作用吗？
在我的应用程序中，我想利用批处理过程中内存中的缓存。
浏览 0提问于2015-12-29得票数 2
回答已采纳
我查阅了关于如何发送数据的不同问答，以及在Android系统活动之间发送数据的最佳方式是什么，但我找不到一个答案，如果数据敏感，什么是最安全的方法。
在android中在活动之间发送数据的最佳方法：
这些方式中哪一种是发送数据的最安全和最有效的方式？
浏览 4提问于2017-01-14得票数 0
回答已采纳
背景：，我编写了一个简单的火花结构化蒸汽应用程序，可以将数据从卡夫卡转移到S3。发现为了准确地支持-一旦保证火花创建_spark_metadata文件夹，最终变得太大，当流式应用程序运行了很长时间，元数据文件夹变得如此大，我们开始得到OOM错误。我想摆脱星火结构化流的元数据和检查点文件夹，并自己管理偏移。
我们如何管理火花流中的偏移：，我已经使用val offsetRanges = rdd.asInstanceOfHasOffsetRanges.offsetRanges来获得Spark结构化流中的偏移量。但想知道如何获得偏移和其他元数据来管理检查点，我们自己使用Spark结构化流。您有实现检查
浏览 2提问于2020-06-17得票数 1
回答已采纳
2回答
我们知道Flink是一个真正的实时流处理引擎，它可以在记录到达时处理记录，我们也知道火花流是一个微批量流处理引擎。
然而，我们也知道星星之火发布了结构化的流媒体，它怎么样？它是否也是一个真正的实时流处理引擎，就像Flink一样，它可以在记录到达时立即处理记录，而不是微批处理，或者仍然使用微批处理模式？
浏览 3提问于2017-06-27得票数 3
回答已采纳
1回答
我对数据流编程模型很陌生，对于数据流如何在窗口流过程中存储中间状态，我有几个问题。假设我白天在窗口，然后执行聚合。当一个新事件出现时，它需要访问该窗口和组中的所有数据。
这些数据是存储在内存中、磁盘上、GCS中还是完全不同的地方？
中间数据的数量会影响作业所需的机器数量吗？
当窗口关闭时，数据会发生什么变化？
如果我正在执行一项操作，例如求和，它不要求将所有数据存储在中间状态，是否有一种方法可以告诉数据流只存储上次更新的结果？
浏览 1提问于2016-06-30得票数 0
回答已采纳
1回答
请帮助我理解为什么在D:/DataStore_ehCache.But中没有文件或任何东西，在执行程序时没有异常或错误。
Profit profit = new Profit();
profit.setId(1);
profit.setAmt("1000");
profit.setLastupdate(new Date());
DiskStoreConfiguration disk =new DiskStoreConfiguration();
disk.setPath("D:/DataStore_ehCache");
Configuration cacheMa
浏览 1提问于2014-07-08得票数 1
回答已采纳
3回答
我使用的是JPA3、Hibernate 4和MySQL 5。我的测试代码看起来像...
@Repository
public class TestRepositoryImpl implements TestRepository {
@PersistenceContext
private EntityManager em;
@Override
@Transactional
public void insertBulk() {
Item it;
for(int i= 0; i<1000;i++) {
it = new Item();
浏览 0提问于2014-01-22得票数 6
由于RDD的血统，RDD本质上是容错的。但是，如果一个应用程序有100多个操作，那么就很难对所有这些操作进行重构。有办法存储中间数据吗？据我所知，存在持久化()/cache()来保存RDD的选项。但是，它们是否足以保存中间数据呢？检查指向会是一个选择吗？另外，是否有一种方法可以在指向RDD时指定存储级别？(如内存或磁盘等)
浏览 3提问于2016-05-06得票数 1
回答已采纳
假设有一个公共项目publicProject。你也有自己的私人privateProject项目。
是否有一种方法可以在每次privateProject收到新推送时自动触发CI/CD管道wihin publicProject？注意，尽管publicProject是公开的，但是没有办法修改publicProject中的管道--因此不可能触发。
这有可能吗？
浏览 2提问于2022-02-15得票数 1
带有Hibernate 5.6的Spring 5.3：
在将Hibernate条件API重写/迁移到JPA CriteriaBuilder之后，我遇到了一些性能问题，findByCriteria方法在我的应用程序启动期间运行了大约15K次。通常，使用条件API可能需要一分钟的时间，但是当我运行JPA代码时，直到~9k迭代(大约需要10秒)为止，一切都很好。但是从~10k开始，执行速度下降到每秒大约80个查询。
知道为什么吗？我的代码如下所示：
public List<ENTITY> findByCriteria(final Map<String, Object> cri
浏览 14提问于2022-09-21得票数 0
回答已采纳
1回答
我是新来的火花流节目。如何显示和存储结果，然后新结果取决于新的DStream和这个旧的结果。spark版本=1.6.0 kafka版本2.10和java
请帮帮忙。
浏览 0提问于2018-05-07得票数 1
回答已采纳
我正在使用apollo和vue (nuxt)构建一个应用程序，我想知道是否必须将已经获取的页面保存在我的商店中，如果我想让它超级快速，无论何时用户返回到他已经访问的页面。 现在我有一个部分，在显示页面之前，我获取了一些apollo查询。收到数据后，我可以将其保存到存储中，以便在有人返回该页面时准备好内容。 但也许Apollo正在用它的智能缓存行为为我做这项工作？ 感谢你在这个问题上的一些意见。干杯
浏览 14提问于2019-09-02得票数 0
回答已采纳
2回答
一个场景，需要使用rest webservice，它提供一个巨大的文件作为流输出，反之亦然，需要处理流并直接写入文件而不是内存。服务： @RequestMapping(value = "downloadFile", method = RequestMethod.GET)
public StreamingResponseBody getSteamingFile(HttpServletResponse response) throws IOException {
response.setContentType("application/octet-stream&#
浏览 50提问于2020-04-08得票数 0
我是通过理论实体管理导入一种csv数据，然而，我有一个循环，执行批处理，因为它是在文档级别上提到的。
$validationErrors = [];
foreach($data as $itrationNumber => $item) {
/** @var User|null $user **/
$user = $this->em->getRepository(User::class)->findOnBy(['email' => $item['email']]);
$user = ($user) ? $use
浏览 2提问于2020-09-28得票数 0
当我们想要用大量数据集“保存”内存时，使用zero_grad()方法。这里已经有了一个答案：。
渐变用于在后支柱过程中更新参数。但是，如果我们通过将梯度设置为0来删除梯度，如何在反向传播过程中进行优化？在有些模型中，我们使用这种方法，并且仍然在进行优化，这是怎么可能的？
浏览 6提问于2022-02-20得票数 -1
1回答
我希望使用Core数据作为数据库服务器上较大数据集的缓存。并不是所有的数据都在内存中。
在思考这个问题时，又想到了两个问题：
可以与内存中的持久存储一起使用故障处理(例如，对1-n个关系)，如果是的话，如何捕获故障触发？核心数据托管对象上下文具有严重的间隔。这也适用于内存存储吗？
或者我应该为此使用一个NSAtomicStore？
浏览 5提问于2009-11-07得票数 6
回答已采纳
我正在使用twitter流函数，它提供了一个流。我需要使用星火writeStream函数，如：
// Write key-value data from a DataFrame to a specific Kafka topic specified in an option
val ds = df
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.writeStream
.format("kafka")
.option("kafka.bootstrap
浏览 2提问于2019-11-13得票数 1
回答已采纳
1回答
每当我通过dcef3实例(TChromium)导航到dcef3时，“欢迎”位总是出现在顶部。我能以某种方式关掉它吗？如果我使用Chrome或火狐浏览器访问maps.google.com，它就不会出现。
我认为这可能与cookie有关，所以尝试将CefPersistSessionCookies设置为true，但没有运气。也许我需要用TChromium设置一个选项属性，或者处理其他事件？我没有对TChromium的默认值做任何其他更改。
浏览 3提问于2014-05-15得票数 2
回答已采纳
1回答
我是Apache的新手，我有几个基本的问题，在阅读火花材料时，我无法理解。每一种材料都有自己的解释风格。我正在使用PySpark朱庇特笔记本在Ubuntu上练习。
根据我的理解，当我运行以下命令时，testfile.csv中的数据被分区并存储在相应节点的内存中。
rdd1 = sc.textFile("testfile.csv")
我的问题是，当我运行以下转换和操作命令时，rdd2数据将存储在哪里。
1.它是否存储在记忆中？
rdd2 = rdd1.map( lambda x: x.split(",") )
rdd2.count()
我知道rdd2中的数据在关
浏览 2提问于2016-07-26得票数 6
回答已采纳
下面是测试用例和结果： drop table if exists test1;
drop table if exists test2;
drop trigger if exists test1_tr on test1;
drop function if exists tf_test1;
create table test1 (name varchar(8) not null);
create table test2 (name varchar(8) not null);
\echo create trigger function tf_test1
CREATE OR REPLACE FUN
浏览 52提问于2021-08-27得票数 0
Dataset<Row> Data_one = spark.read().format("csv").option("header","true").load("src/main/resources/exams/students.csv");
System.out.println("Before persist");
Data_one.show();
Data_one.persist(StorageLevel.MEMORY_ONLY());
浏览 13提问于2020-02-10得票数 0
我想了解一下flush方法在下面的例子中到底做了什么：
for (int i = 0; i < myList.size(); i++) {
Car c = new Car( car.get(i).getId(),car.get(i).getName() );
getCurrentSession().save(c);
if (i % 20 == 0)
getCurrentSession().flush();
}
这是否意味着在迭代20之后，缓存被刷新，然后所持有的20个内存对象实际上被保存在数据库中？
有人能给我解释一下当条件为真时会发生什么吗？
浏览 1提问于2010-07-18得票数 29
回答已采纳
2回答
总之，我有一个在数据存储中写东西的代码，在我调用检索我刚刚写到数据存储中的数据之后。我的第二次呼叫总是失败！但是，如果我再次调用，它将正常工作并找到数据！它真的很有线缆！代码在1.7.1上运行得很好，我在升级到1.7.2后开始发生这种情况(不能百分之百确定，因为我同时还做了一些其他更改，但我想我也提到了这一点)。下面是我编写数据的代码：
PersistenceManager pm = PMF.get().getPersistenceManager();
pm.evictAll();
Key userKey = KeyFactory.createKey(User.clas
浏览 0提问于2012-10-02得票数 0
1回答
我们有一些数据(数百万)在蜂箱表中，每天都有。第二天，一旦熬夜摄入完成，不同的应用程序就会查询我们的数据(使用sql)。
我们使用这个sql并调用spark。
spark.sqlContext.sql(statement) // hive-metastore integration is enabled
这导致火花驱动程序占用过多内存，我们是否可以使用火花流(或结构化流)以管道方式流，而不是收集驱动程序上的所有内容，然后发送给客户端？
我们不想(在典型的流应用程序中)立即发送数据，而是希望在客户要求(拉)数据时向客户发送流数据。
浏览 1提问于2020-07-12得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
