在tf.keras中使用Horovod时，如何从检查点恢复？
在tf.keras中使用Horovod时，可以通过以下步骤从检查点恢复：
- 导入必要的库和模块：
import tensorflow as tf
import horovod.tensorflow.keras as hvd- 初始化Horovod：
hvd.init()- 配置TensorFlow会话：
config = tf.ConfigProto()
config.gpu_options.visible_device_list = str(hvd.local_rank())
tf.keras.backend.set_session(tf.Session(config=config))- 定义模型：
model = tf.keras.models.Sequential()
# 添加模型层- 编译模型：
optimizer = tf.keras.optimizers.Adam(0.001 * hvd.size())
optimizer = hvd.DistributedOptimizer(optimizer)
model.compile(optimizer=optimizer, loss='categorical_crossentropy', metrics=['accuracy'])- 定义检查点回调函数：
checkpoint_dir = './checkpoints'
if hvd.rank() == 0:
os.makedirs(checkpoint_dir, exist_ok=True)
checkpoint_file = os.path.join(checkpoint_dir, 'checkpoint.h5')
checkpoint_callback = tf.keras.callbacks.ModelCheckpoint(checkpoint_file, save_weights_only=True)- 加载检查点（如果存在）：
if os.path.exists(checkpoint_file):
model.load_weights(checkpoint_file)- 训练模型：
model.fit(x_train, y_train, callbacks=[checkpoint_callback], ...)通过以上步骤，可以在使用Horovod进行分布式训练时，从检查点恢复模型。注意，每个训练节点都会保存自己的检查点，但只有rank为0的节点会加载检查点。这样可以确保在分布式训练中，只有一个节点负责保存和加载检查点。
推荐的腾讯云相关产品：腾讯云AI加速器、腾讯云弹性GPU、腾讯云容器服务等。你可以通过访问腾讯云官方网站获取更多关于这些产品的详细信息和介绍。
相关·内容
2回答
我正在尝试理解Tensorflow镜像策略和Horovod分发策略之间的基本区别。 从文档和源代码调查中，我发现Horovod (https://github.com/horovod/horovod)使用消息传递协议在多个节点之间进行通信。具体地说，它使用了MPI的all_gather all_reduce。 根据我的观察(我可能是错的)，镜像策略也使用all_reduce算法(https://github.com/tensorflow/tensorflow/tree/master/tensorflow/contrib/distribute)。 它们都使用数据并行、同步训练方法。所以我有点困
浏览 16提问于2019-03-06得票数 6
我已经开始探索TensorFlow库，并尝试使用MNIST数据的图像分类。我希望模型在培训阶段结束后存储在一个文件中，这样我就可以在需要时使用它。我检查了，它告诉我们如何将值从TensorFlow保存到任何文件中，并读取它。到目前为止，我能够按照链接中的建议使用泡菜将脚本中的一些变量保存到文件中。但是，我无法理解需要在文件中保存什么来存储模型的当前状态以供以后使用。拜托，有人可以用例子来解释这个部分，比如存储模型和加载模型。
浏览 13提问于2017-02-21得票数 0
回答已采纳
我使用最新的tf.keras TF2.0a创建、编译并安装了一个模型。现在，我想将它保存为pb，以便稍后用于推理。
我用：
import time
saved_model_path = "./models/pb/experimental/{}".format(int(time.time()))
tf.keras.experimental.export_saved_model(model, saved_model_path)
saved_model_path
正如在point下面很好地描述为saved_model
model是我以前创建/编译/安装的模型。
现在我的问题
e
浏览 0提问于2019-05-22得票数 1
回答已采纳
我必须聚合不同实体的事件流的计数/总和。从kafka接收事件日志( json _ str)，并填充映射实体名作为键，值是选择性属性的计数作为json_ str。 MapState sourceAggregationMap = getRuntimeContext().getMapState(sourceAggregationDesc)； 对于每个事件流，重新填充值。 问题是，每当作业停止(失败)/cancelled时，当作业重新启动时，映射状态不会重新初始化/恢复。计数再次从0开始。 使用Apache flink 1.6.0 state.backend: rocksdb
浏览 25提问于2019-01-11得票数 1
回答已采纳
我正在试图解决一个CouchDB被回滚的问题，导致PouchDB在将来出现。我希望找到一种方法来检测这种情况，并在这种情况发生时强制PouchDB销毁和重新加载。
是否有一种方法可以要求PouchDB提供它当前的拉力seq号？我根本找不到这方面的任何文件。我的google不够强大。
到目前为止，我唯一的想法是观看sync.on(变更)提要，并在每次拉时记录seq号。然后，在app重新加载时，以ajax https:/server/db/_changes?descending=true&limit=1的形式运行它，并验证它返回的seq号是否高于我存储的seq编号。如果存储的seq较高，则
浏览 1提问于2018-04-23得票数 0
回答已采纳
1回答
我使用的是在从docker-flink安装的ECS中运行的Flink。我已经通过flink-conf.yaml中的state.checkpoints.dir到S3启用了外部检查点到AWS S3。 现在，根据Flink文档here，如果我们想在出现故障的情况下从检查点恢复，我们必须使用bin/flink run -s :checkpointMetaDataPath [:runArgs]，但我使用FLINK_HOME/bin standalone-job.sh start-foreground。所以我不能弄清楚我的Flink作业在失败的情况下如何从外部化的检查点恢复。 我们真的需要一些从检查点恢
浏览 35提问于2020-04-03得票数 0
回答已采纳
当我们将数据库从SQL2K12 stanadlone移动到新的服务器/实例时，我在下面看到了错误消息。
最后未完成的目标: 20800，avgWriteLatency 30平均吞吐量:69.12MB/秒，I/O饱和: 243550，上下文切换398948 FlushCache:清除579723个bufs，388331写入65526 ms (避免125170个新的脏bufs)的db 19:0
我不知道这是否指出了斯泰格的问题。
现在需要关注的是，在我们进行迁移之前，我在这个实例上有类似的错误消息，但是它们类似于半小时后(2小时或15分钟后)。没有固定的图案。
但是在迁移之后，从我们迁移的那一天开
浏览 0提问于2018-03-08得票数 1
Given，
我有一个Flink作业，它从ActiveMQ源代码读取&写到mysql数据库，键控在标识符上。我每隔一秒就为这项工作启用检查点。我将检查点指向一个Minio实例，验证检查点是否与jobid一起工作。我部署这个任务是一个Openshift (在下面的Kubernetes)--我可以在需要的情况下扩大/减少这个任务。
问题
在部署(滚动)作业或由于错误/错误导致作业下降时，如果ActiveMQ中有未使用的消息或Flink中的未确认消息(但写入数据库)，则当作业恢复(或部署新作业)时，作业处理已经处理消息，从而导致在数据库中插入重复的记录。
问题
检查点不应该帮助工作从原来
浏览 1提问于2020-03-05得票数 1
2回答
我使用来自的方法2，我使用检查点，当我必须更改代码，并重新部署我的代码时，有时检查点会抛出异常，如果由于某种原因，我必须删除检查点目录，我如何重用检查点目录文件夹来从kafka获取消息，我认为检查点目录存储了kafka偏移量。
浏览 5提问于2016-03-31得票数 1
在恢复时间或服务器重新启动时，PostgreSQL从pg_control文件中找到最后一个检查点记录，并从这里开始恢复。
在重播WAL目录中的WAL段时，PostgreSQL将数据文件与每个checkpoint_timeout (5分钟)或max_wal_size边界上的WAL记录同步。
它用名为"restartpoint“的LSN设置更新每个检查点边界上的pg_control文件，以便在下次重新启动恢复时，可以从该点开始。
我的理解正确吗？
浏览 5提问于2022-08-21得票数 0
1回答
当我使用flink sql处理数据时。
重新启动应用程序，sum结果不会保存在校验点。它仍然从1开始。
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StateBackend stateBackend = new FsStateBackend("file:///D:/d_backup/github/flink-best-practice/checkpoint");
env.enableCheckpointing(1000 * 60);
en
浏览 0提问于2018-12-28得票数 0
正在将和MPI组合成一个用于分布式深度学习的包装器，例如TensorFlow。我以前没有听说过NCCL，我正在研究它的功能。以下是NVIDIA网站上关于NCCL的说明：
NVIDIA集体通信库(NCCL)实现了对NVIDIA GPU性能进行优化的多GPU和多节点集体通信原语。
从关于NCCL的中，我了解到NCCL是通过PCIe、NVLink、原生Infiniband、以太网工作的，它甚至可以通过RDMA检测GPU在当前硬件拓扑中是否有意义，并透明地使用它。
所以我想问为什么在霍洛沃德需要MPI？据我所知，MPI还被用来通过also范式有效地交换分布式节点之间的梯度。但据我所知，NCCL
浏览 2提问于2018-11-27得票数 11
我正在学习TensorFlow文档中的“神经机器翻译与注意”教程，但无法知道如何将模型保存为SavedModel文件。
从文档中可以看出，我可以很容易地保存检查点，但是afaik在与其他应用程序集成时并不十分有用。有人知道如何保存整个“模型”，即使他们没有使用tf.keras.Model
博士：
浏览 1提问于2020-11-05得票数 1
回答已采纳
当我尝试创建检查点时，为什么perforce会出现此错误？我是否可以仅从检查点文件和日志文件恢复整个数据库？我做错了什么，这是如何工作的？为什么perforce用户要指导一本巨型的书，而网上却没有视频教程？
浏览 0提问于2020-09-10得票数 0
1回答
我想知道flink.In单机模式的详细故障恢复过程，我猜有一些步骤，比如TaskManager故障，首先检测故障，所有任务停止处理，然后重新部署任务。然后从HDFS下载检查点，每个操作员加载状态。加载完成后，源端将继续发送数据。我说的对吗？有没有人知道正确和详细的恢复过程？
浏览 3提问于2019-04-03得票数 0
1回答
在TaskManagers( TM )的HA上下文中，当TM发生故障时，将从JobManager(JM)最近的故障检查点恢复新的TM。
假设我们在给定时间t有3个TM (tm1, tm2, & tm3)，每个人的检查点(Cp)都在cp1。它们之间的所有TMs广播数据。
现在tm2宕机了，JM带来了带有cp1检查点的tm2'作为HA的一部分。到了t+x一个新的TM被提出的时候，其他人也发展到了cp2。
tm1和tm3作为cp2的一部分广播的数据是如何在tm2'上重放的
浏览 11提问于2020-07-31得票数 0
1回答
Apache通过从检查点恢复作业来保证在故障和恢复时处理一次，检查点是分布式数据流和操作符状态的一致快照(Chandy-Lamport算法用于分布式快照)。此保证在故障转移时恰好是一次。
在正常的集群操作中，Flink如何保证只处理一次，例如给定从外部源读取的Flink源(例如Kafka)，Flink如何保证事件从源读取一次？事件源和Flink源之间是否存在某种类型的应用程序级别？另外，Flink如何保证事件准确地从上游运营商到下游运营商之间只发生一次？对于接收到的事件，这也需要任何形式的附加条件吗？
浏览 3提问于2020-09-30得票数 3
回答已采纳
2回答
当Sql Server发出检查点时，它是否会阻止所有其他操作，直到检查点完成？
如果我理解正确的话，当检查点发生时，服务器应该写入所有脏页。
当它完成时，它会将检查点写入事务日志，因此在任何失败的情况下，它将只处理从该时间点开始的事务(或在检查点时已经启动的事务)。
当检查点正在运行时，sql server如何防止某些非脏页变脏？
它是否会在检查点完成之前阻止所有写入？
浏览 0提问于2012-10-16得票数 1
回答已采纳
我在CNTK中调用trainer.restore_from_checkpoint时遇到了以下异常。
“此”函数与从检查点恢复的函数不等效(同构)。
我的恢复代码如下。它们与创建训练器和通过trainer.dnn保存trainer.save_checkpoint("trainer.dnn")的结构相同，如所提到的。
def evaluate(reader, model):
criterion = create_criterion_function(model)
criterion.replace_placeholders({criterion.placeh
浏览 0提问于2017-01-21得票数 1
我们正在使用Flink Kinesis Consumer将Kinesis流中的数据消费到我们的Flink应用程序中。 KCL库使用DynamoDB表来存储最后成功处理的Kinesis流序列号。因此，下次应用程序启动时，它将从中断的位置恢复。 但是，Flink Kinesis Consumer似乎没有维护任何这样的序列号。在任何持久化存储中。因此，我们需要依靠ShardIteratortype (trim_horizen、latest等)来决定在应用程序重启时从哪里恢复Flink应用程序处理。 一种可能的解决方案是依赖于Flink检查点机制，但只有当应用程序在失败后恢复时才起作用，而不是当应用
浏览 16提问于2019-02-22得票数 3
1回答
我的flink作业使用FlinkKafkaConsumer010从kafka consumer读取数据，并使用CustomBucketingSink接收到hdfs。我们有一系列的转换kafka -> flatmaps(2-3个转换) -> keyBy -> tumblingWindow(5 mins) -> Aggregation -> hdfsSink。我们有平均300万/分钟事件的kafka输入，高峰时间大约2000万/分钟的事件。检查点持续时间和两个检查点之间的最小停顿是3分钟，我使用的是FsStateBackend。
以下是我的假设：
Flink从kafk
浏览 0提问于2020-03-12得票数 0
为了学习的目的，我试图在检查点上做一些实验/测试。
但是我的选择有限，我只能看到内部的运作。我正试着从套接字上读。
val lines: DataFrame = spark.readStream
.format("socket")
.option("host", "localhost")
.option("port", 12345)
.load()
并使用它执行一些需要检查点的状态操作。
Q1。使用检查点位置作为我的本地系统时，它无法读取检查点，并且会出现错误。
This query does not suppor
浏览 3提问于2021-01-25得票数 1
回答已采纳
1回答
我将增量检查点与RocksDB结合使用，并将检查点保存到远程目标(在我的示例中为S3)。如果有人删除作业管理器服务器(检查点协调器在其中运行)并重新安装它，会发生什么情况？通过失去检查点协调器，我还失去了从检查点恢复状态的选项？因为据我所知，协调器持有检查点的所有引用。
浏览 15提问于2019-04-10得票数 0
回答已采纳
在postgresql.conf版本13的PostgreSQL文件中，archive_cleanup_command注释以以下方式解释了该命令：
#archive_cleanup_command = '' # command to execute at every restartpoint.
文档和没有提到“重新开始点”。这引起了以下问题：
，什么是重新开始点？，例如: restartpoint仅仅是一个检查点的相同词吗？这两者的意思是完全一样的吗？
例如，什么时候创建了一个重新启动点？：如果重新启动点只是一个检查点，那么每5分钟就会创建一个检查点，或者
浏览 2提问于2020-11-16得票数 5
回答已采纳
我有多个Kafka主题(多租户)，并且我根据主题的数量多次运行相同的作业，每个作业消耗来自一个主题的消息。我已经将文件系统配置为状态后端。 假设有3个作业在运行。检查点在这里是如何工作的？这3个作业是否都将检查点信息存储在同一路径中？如果任何作业失败，该作业如何知道从何处恢复检查点信息？我们过去常常在向flink群集提交作业时提供作业名称。这跟它有什么关系吗？一般而言，Flink如何区分作业及其检查点信息，以便在作业出现故障或手动重新启动作业时进行恢复(无论作业相同还是不同)？ Case1:如果作业失败，会发生什么情况？ Case2:如果我们手动重新启动作业会发生什么？ 谢谢
浏览 33提问于2020-07-16得票数 0
回答已采纳
1回答
在本地运行多个mesos从站
、、、、
我试图在本地运行一个测试集群，遵循本指南
目前，我可以在localhost:5050上运行一个主服务器，在默认端口5051上运行一个从服务器(使用从id (如S0) )。但是，当我试图在另一个端口启动另一个从站时，它重新注册为S0，主控制台只显示了一个激活的从机。有人知道我如何启动另一个奴隶S1吗？谢谢!
浏览 2提问于2015-04-02得票数 0
回答已采纳
1回答
在Akka流应用程序中，如果应用程序崩溃/节点崩溃，我就有丢失正在运行的消息的风险。
我们可以做些什么来确保我们不会丢失消息，并且一旦应用程序回来就会播放它们。
在我的应用程序中，mapAsyns执行IO调用和一些次要的CPU绑定任务。
在纯Akka应用程序中，我将选择持久队列或事件源。但是，如果是Akka流，因为我无法访问底层队列，我如何确保流中没有消息丢失？
浏览 0提问于2016-12-12得票数 1
回答已采纳
我已经创建了一个带有一堆变量的模型(模型A)。我计划使用Model A中的一些层在新模型(Model B)上与Model A进行迁移学习。但是，Model B与Model A具有相同的体系结构，因此我不能在运行model B之前从Model A加载所有变量，否则命名等会出现错误。因此，我正在尝试创建一个新的ckpt文件，它只存储我想要从Model A获得的权重。然后，我将使用这个新的ckpt文件加载到Model B中。我有以下内容：
sess = tf.Session()
saver = tf.train.import_meta_graph('ModelA.ckpt.meta'
浏览 0提问于2017-03-30得票数 3
1回答
在使用apache-时，我试图对一些流数据应用"reduceByKeyAndWindow()“转换，并得到以下错误：
pyspark.sql.utils.IllegalArgumentException: requirement failed: The checkpoint directory has not been set. Please set it by StreamingContext.checkpoint().
是否需要设置检查点目录?
如果是的话，最简单的设置方法是什么？
浏览 4提问于2015-11-17得票数 4
回答已采纳
Jupyter Notebook商店是否会更改.ipynb文件中的历史记录？
我使用Jupyter Notebook开发了一个API客户端。如果我对敏感信息进行硬编码，那么一遍又一遍运行代码的速度要快得多。我担心，即使我在添加到公共存储库之前清理代码，也可能会从.ipynb文件的新副本中追溯到旧的更改。
示例：
g = GenieClient('eb243592-faa2-4ba2-a551q-1brsf565c889') # (an invalid key for demonstration)
# g = GenieClient(input('OpsGenie API
浏览 4提问于2018-11-28得票数 1
1回答
我有一个星火流作业运行在一个集群(Spark1.6)上，它是S3的检查点。当我刚开始工作时，我可以看到“流”选项卡。但是，当我从检查点重新启动作业时，流选项卡就会消失。作业仍然作为流作业工作，我看到批处理出现在已配置的批处理间隔中。见下文。
如果清除检查点数据，选项卡将返回。我怀疑流选项卡在从检查点重新启动时没有正确注册。
我看了看。当应用程序状态从检查点反序列化时，是否可能不调用此流？
有人知道怎么解决这个问题吗？
浏览 4提问于2016-04-18得票数 2
回答已采纳
这是一个关于Tensorflow对象-检测API的更普遍的问题。
我正在使用这个API，更确切地说，我微调了一个模型到我的数据集。根据API的描述，我使用model_main.py函数从给定的检查点/冻结图重新训练模型。
然而，我不清楚微调在API中是如何工作的。最后一层的重新初始化是自动发生的还是我必须实现类似的东西？在README文件中，我没有找到任何关于这个主题的提示。也许有人能帮我。
浏览 0提问于2019-05-06得票数 9
回答已采纳
我使用Yarn来运行Flink作业。对于每个Flink作业，我都要创建一个检查点。
我提交一个在我的Yarn集群中运行的Flink作业。我有一个轮询作业，它检查一个作业是否在Yarn上失败并重新启动它。当作业再次提交时，Yarn为该Flink作业创建一个新的application_id。如何配置重新提交的Flink作业以使用重新启动的Flink作业的检查点。
我已将conf state.savepoints.dir = hdfs://localhost:9000/checkpoint/设为flink-conf. have‘
创建Flink作业时，streamExecutionEnvironme
浏览 0提问于2018-05-22得票数 0
回答已采纳
1回答
我正在运行tensorflow object_detection模型的model_main.py。我想知道哪个.py文件包含代码，模型在哪里检查'mode_dir‘中的现有检查点，以及模型如何知道从保存的最新检查点加载/启动。 我检查了model_lib.py，但还不能弄清楚。
浏览 13提问于2019-06-15得票数 0
在tensorflow中，我需要从inception_v3预培训模型加载权重，以便在以下代码中使用：
with tf.variable_scope(scope, "InceptionV3", [images]) as scope:
with slim.arg_scope(
[slim.conv2d, slim.fully_connected],
weights_regularizer=weights_regularizer,
trainable=False):
浏览 0提问于2018-01-29得票数 1
回答已采纳
在进行更改时，将数据页从数据库数据文件中提取到缓冲区中，并对缓冲区页进行更改，并将更改写入事务日志。先写日志记录。在简单恢复模式下，缓冲区中的更改将应用于检查点处的数据库数据页。检查点从事务日志中刷新记录。因此，不能在此模式下进行事务日志备份。类似地，对于尚未进行完整备份的完整/批量恢复模型。
问题:一旦进行了第一次完全备份，对数据页的更改将如何以及何时写入完整/批量恢复模型中的数据库数据文件？
浏览 0提问于2018-10-15得票数 0
回答已采纳
我想在训练期间每5小时保存一次tensorflow变量。 因此，根据Tensorflow Saver doc，我构造了带有参数keep_checkpoint_every_n_hours=5的saver = Saver，并为学习的每一步调用saver.save()。因此，我所期望的是，saver模块以某种方式检测开始训练后经过的时间，并每5小时保存一次模型，而不是在调用它的每个步骤中实际保存模型。 下面简化了我如何使用该功能。 sess = tf.Session()
model = Model(sess)
saver = tf.train.Saver(max_to_keep=5,
浏览 35提问于2019-05-13得票数 1
回答已采纳
我使用MQTT使用者作为flink作业的数据源。我想知道如何将数据偏移保存到检查点，以确保在flink集群发生故障后重新启动时不会丢失任何数据。我看到了很多介绍apache如何管理kafka消费者补偿的文章。有谁知道apache是否有自己的功能来管理MQTT使用者？谢谢。
浏览 24提问于2022-01-24得票数 0
我正在尝试恢复数据库中的数据。如何在没有任何数据库备份的情况下从WAL恢复数据？我可以从WAL日志生成一个备份文件吗？
浏览 0提问于2017-08-14得票数 1
回答已采纳
我们有一个场景，在这个场景中，我们必须将一些值保存到检查点中，并在故障恢复/应用程序重新启动期间将其检索回来。
我们跟踪了一些事情，比如ValueState，ValueStateDescriptor仍然不起作用。
我们不能将它外部化为DB，因为它可能会导致一些性能问题。任何导致这一点的结果都将有助于使用检查点。如何放置并从检查站返回？
浏览 10提问于2022-01-19得票数 0
1回答
我有一个组件可以将他的结构写到一个文件中。问题是，他写入的数据需要在任何系统故障时保持一致。所以我需要物理地写数据。问题是组件没有这样的选项，从组件数据创建文件保存函数需要花费一些时间，并且会使程序变得复杂。
问题是:如果我使用组件(例如:ComponentX->WriteToFile(文件名))写入数据，我是否可以使用Handle =OpenFile(文件名)，然后使用FlushFileBuffers(句柄)来确保数据的一致性？否则这个把戏就不管用了？
我认为这可能是可行的，因为在OpenFile中，系统可能会使用缓存中已有的句柄，刷新它将导致保存上一次操作(组件文件保存)中的缓存数据
浏览 3提问于2012-01-29得票数 1
回答已采纳
1回答
我已经设置了一个具有2 JobManagers和3 TaskManagers的Flink 1.2独立集群，我使用JMeter通过产生卡夫卡消息/事件来对其进行负载测试，然后对这些消息/事件进行处理。处理作业在TaskManager上运行，通常需要15K事件/秒。
该作业设置了EXACTLY_ONCE检查点，并将状态和检查点持久化到S3。如果我关闭运行作业的TaskManager，则需要几秒钟时间，然后在另一个TaskManager上继续作业。作业主要记录事件ids，这些事件ids是连续整数(例如，从0到1200000)。
当我检查TaskManager上的输出时，我关闭了最后一个计数，例如50
浏览 1提问于2017-04-16得票数 5
回答已采纳
我正在试验我的新Flink集群(3台不同的机器->1个作业管理器，2->任务管理器)，使用RocksDB作为状态后端，但是我得到的检查点行为有点混乱。 更具体地说，我设计了一个简单的WordCount示例，我的数据源是netcat。当我提交作业时，作业管理器将其分配给一个随机的任务管理器(也不复制)。我提供了一些单词，然后我杀死了当前正在运行的任务管理器。一段时间后，作业在另一个任务管理器中重新启动，我可以提供一些新单词。令人困惑的是，即使我杀死了第一个任务管理器，它的状态也会保留下来。 据我所知，RocksDB在运行的任务管理器的本地目录中维护它的状态，所以我所期望的是，当第一
浏览 32提问于2021-04-02得票数 0
回答已采纳
1回答
我试图从聚合原则的角度来理解spark streaming。Spark DF是基于小批量的，并且在特定时间窗口内的小批量上进行计算。
假设我们有数据以-
Window_period_1[Data1, Data2, Data3]
Window_period_2[Data4, Data5, Data6]
..
首先对Window_period_1进行计算，然后对Window_period_2进行计算，如果我需要使用新的传入数据和历史数据，比如Window_period_new和Window_period_1和Window_period_2的数据之间的分组函数，我该怎么做？
另一
浏览 1提问于2016-05-06得票数 4
3回答
SQLServer2008在一台PC上。当我们升级到SQLServer2008时，开始出现一个ODBC错误：
State: 37000,Native: 155,Origin: [Microsoft][Sql Natize Client][SQL Server]
'TRUNCATE_ONLY' is not recognized as a BACKUP option.
现在，这是错误消息已经开始成为一个常见的事件。我有麻烦了。
如何降低它的等级？请帮帮我
浏览 0提问于2009-06-15得票数 0
我设置了spark.streaming.receiver.writeAheadLog.enable=false，我的历史服务器确认属性已经设置。然而，我仍然看到这个错误：
16/11/28 15:47:04 ERROR util.FileBasedWriteAheadLog_ReceivedBlockTracker: Failed to write to write ahead log after 3 failures
我从spark users线程中推测，必须启用HDFS append支持才能让WAL与HDFS一起工作。我的安装没有启用这个HDFS功能，所以我想在Spark中禁用WAL。
浏览 1提问于2016-11-29得票数 0
1回答
我在类似的用例中使用了transform方法，如Transform 节中所描述的
spamInfoRDD = sc.pickleFile(...) # RDD containing spam information
# join data stream with spam information to do data cleaning
cleanedDStream = wordCounts.transform(lambda rdd: rdd.join(spamInfoRDD).filter(...))
我的代码如下：
sc = SparkContext("local[4]",
浏览 2提问于2015-06-17得票数 4
在处理过程中添加了一些托管状态之后，我们发现尽管在RocksDb中使用了增量检查点，但检查点的大小和持续时间却出现了令人担忧的增长。
为了隔离这个问题，我们用源、映射操作符和接收器创建了简单的拓扑。
源在内存中创建具有每秒一个事件吞吐量的任意数量的事件。每个事件都有唯一的id，用于分区流(使用keyBy操作符)，并通过映射函数将大约100 to的值添加到托管状态(使用的ValueState)。然后，简单地将事件传递到不起任何作用的接收器中。
使用上面描述的设置，我们发送了1200个事件，检查点间隔和最小暂停设置为5秒。当事件以恒定的速度和相同的状态出现时，我们预计检查点的大小或多或少是恒定的。
浏览 0提问于2018-10-31得票数 1
回答已采纳
1回答
我正致力于报告的性能分析，这是由于不同的测试用例造成的。我正在使用MS SQL Server 2016开发版
关于一个性能计数器(‘检查点页/秒’)，我理解计数器的含义。在22秒间隔内，图形从每秒400页下降到200页，然后返回到400页。
图中的结果：
1-13秒，值约400。
14-16，下降到200，然后返回到400。
16-38，价值约400
39，降至200，然后返回400。
61，降至200，然后返回400。
84，降至200，然后返回400。
性能计数器‘磁盘写/S’中的值与‘检查点页/秒’相同。
我原以为这些柜台是一条直线(没有峰谷)。
你能解释一下这种行为吗？检查点页/秒从4
浏览 0提问于2018-05-24得票数 4
回答已采纳
我正在运行一份电火花作业，数据流来自卡夫卡。我试图在我的windows系统中复制一个场景，以了解当消费者下降时，当数据不断被输入Kafka时会发生什么。
这就是我所期望的。
生产者启动并生成消息1、2和3。
使用者是在线的，并且使用消息1、2和3。
现在消费者由于某种原因而下降，而生产者产生消息4、5和6等等……
当消费者出现时，我的期望是它应该读到它停下来的地方。因此，消费者必须能够从第4，5，6等信息中读取.
我的电火花应用程序无法实现我预期的目标。下面是我如何创建星火会话的方法。
session.readStream.format("kafka")
浏览 2提问于2020-05-08得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
