标签: batch-processing

我们有5,000个PDF,总共不应超过200GB.他们可能需要全年更新,批量约为1,000.

如我所见,有两条主要路线......

1)通过Tridion发布PDF和相关元数据2)直接导入交付环境并管理Tridion中的PDF元数据

将这些PDF放入CMS的一个令人信服的(商业)原因是让它们投入生产的途径 - CMS =简单 - 非CMS =根本不容易以及它直接为业务提供的控制.

我们当然更愿意管理与二进制项直接相关的元数据,并且还利用组件链接(跟踪使用的位置等)而不是使用"链接"将组件(用于元数据)映射到非CMS控制的二进制项 - 所以在我看来,通过 CMS会更有意义.

现在 - 存在膨胀数据库/阻止发布队列的问题......

其中一些项目可能需要通过工作流程(如果我们通过WebDAV批量上传,我认为我们可以为特定文件夹定义特定的盒式磁带,从而关联不同的模式？).但是 - 使用WebDAV 可能意味着PDF(和历史版本)将存储在数据库中,这可能会有问题.

所以...我们可以将Tridion中的这些链接作为外部链接组件,但我认为这意味着我们不能使用WebDAV(或者我们仍然可以使用带有external_linked PDF的WebDAV - 看起来它没有意义吗？)

我确信在CMS中(或周围)管理大量二进制文件是我们很多人遇到过的,并且非常有兴趣了解其他人是如何处理这种困境的？

谢谢

我需要将文件x.dtsx从位置a复制到位置b.

如果b中已存在x.dtsx,那么我需要将x.dtsx重命名为x_Standby.dtsx然后,将x.dtsx重命名为b后

我当前的代码如下所示:

if exists %1 rename %1 %(should be 1_standy.extension)
xcopy %1 %2

我有一个长期运行的Torque/PBS工作,我想监控输出.但是只有在作业完成后才会复制日志文件.有没有办法说服PBS刷新它？

基于我的上一篇文章,我能够让批处理工作......直到某一点.除了注册路由特定处理程序之外,我还有2个委派处理程序

1. 验证用户
2. 记录

批处理程序通过委托处理程序验证用户并记录请求.当messagehandlerinvoker开始发送子/嵌套请求时,抛出以下异常.

System.ArgumentException was unhandled by user code
  HResult=-2147024809
  Message=The 'DelegatingHandler' list is invalid because the property 'InnerHandler' of 'AuthenticationMessageHandler' is not null.
Parameter name: handlers
  Source=System.Net.Http.Formatting
  ParamName=handlers
  StackTrace:
       at System.Net.Http.HttpClientFactory.CreatePipeline(HttpMessageHandler innerHandler, IEnumerable`1 handlers)
       at System.Web.Http.HttpServer.Initialize()
       at System.Web.Http.HttpServer.<EnsureInitialized>b__3()
       at System.Threading.LazyInitializer.EnsureInitializedCore[T](T& target, Boolean& initialized, Object& syncLock, Func`1 valueFactory)
       at System.Threading.LazyInitializer.EnsureInitialized[T](T& target, Boolean& initialized, Object& syncLock, Func`1 valueFactory)
       at System.Web.Http.HttpServer.EnsureInitialized()
       at System.Web.Http.HttpServer.SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
       at System.Net.Http.HttpMessageInvoker.SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
       at RoutingRequest.Service.Startup.BatchMessageHandler.<>c__DisplayClassd.<PrcoessRequest>b__b(Task`1 m) in C:\CEI\Clients\Footlocker.com\FL - …

我正在使用Spring Batch解析文件，并且出现以下情况：

我正在工作。这项工作必须解析给定文件。由于意外原因（例如断电），服务器发生故障，我必须重新启动计算机。现在，在重新启动服务器之后，我想从停电之前停止的那点恢复作业。这意味着，如果系统现在从10.000读取1.300行，则必须从1.301行开始读取。

如何使用Spring Batch实现此方案？

关于配置：我使用spring-integration在目录下轮询新文件。到达文件后，spring-integration将创建spring批处理作业。另外，spring-batch使用FlatFileItemReader解析文件。

在用PHP编写批处理器时(例如,它显然必须是cron-ed),使用Gearman和简单地存储要在Redis中处理的数据之间有什么实际区别？

到目前为止,我的观察结果是,虽然Gearman能够实时推动工作,因为PHP代码只是间隔运行,使用Redis的定期调度命令似乎或多或少相当.

此外,似乎使用Gearman通过将其绑定到Gearman库的调度生命周期来为应用程序添加不必要的复杂性.

所有这些都说,假设批处理器不会一直运行,那么认为Gearman + PHP没有提供优于Redis + PHP的好处是正确的吗？

我从1.9迁移到2.2并阅读文档我惊讶地发现在批量操作期间不可能进行升级,因为操作不允许选项.

bulkOps.Add(new UpdateOneModel<BsonDocument>(filter, update));
collection.BulkWrite(bulkOps);

应该

options.isUpsert = true;
bulkOps.Add(new UpdateOneModel<BsonDocument>(filter, update, options));
collection.BulkWrite(bulkOps);

这项工作正在进行中,有意或者我遗失了什么？谢谢.

我正在使用一个大数据集,所以我试图使用train_on_batch(或适合epoch = 1)

model = Sequential()
model.add(LSTM(size,input_shape=input_shape,return_sequences=False))
model.add(Dense(output_dim))
model.add(Activation('softmax'))
model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=["accuracy"])

for e in range(nb_epoch):
    for batch_X, batch_y in batches:
        model.train_on_batch(batch_X,batch_y)
        # or
        # model.fit(batch_X,batch_y,batch_size=batch_size,nb_epoch=1,verbose=1,shuffle=True,)

但是当训练开始时,会发生这种情况

(0, 128)
Epoch 1/1
128/128 [==============================] - 2s - loss: 0.3262 - acc: 0.1130

(129, 257)
Epoch 1/1
128/128 [==============================] - 2s - loss: -0.0000e+00 - acc: 0.0000e+00

无论我等了多少个时代,它都没有改变.即使我改变批量大小,也会发生同样的事情:第一批具有良好的值,然后它再次进入"损失:-0.0000e + 00 - acc:0.0000e + 00".

有人可以帮助理解这里发生的事情吗？

提前致谢

我正在使用Apache Camel SQL批量插入过程.

1. 我的应用程序是从Active MQ读取票据,其中包含大约2000张票据.

2. 我已将批次更新为100.

3. 我要解雇的查询如下:

   sql.subs.insertCdr= insert into subscription_logs(master_id,request_type,req_desc,msisdn,amount,status,resp_code,resp_desc,channel,transaction_id,se_mode,be_mode,sub_type,sub_timeleft,srv_name,srv_id,start_date,end_date,operator,circle,country,time_offset,retry_count,user_status,previous_state,se_reqrecvtime,se_respsenttime,be_reqsenttime,be_resprecvtime,cp_id,cp_name,sub_srvname,sub_srvid,msg_senderid,msg_text,call_back_url,call_back_resp,client_ip,se_sysIp,language,cp_callbackurlhittime,action,alert,notification_url,notification_resp) values(:#masterId, :#requestType,:#reqDesc,:#msisdnCdr,:#price,:#status,:#responseCode,:#reason,:#channel,:#transactionId,:#seMode,:#beMode,:#subType,:#subTimeLeft,:#serviceName,:#serviceId,:#subStartDate,:#cdrEndDate,:#operator,:#circle,:#country,:#timeOffset,:#retryCount,:#userStatus,:#previousState,:#seReqRecvTime,:#seRespSentTime,:#beReqSentTime,:#beRespRecvTime,:#cpId,:#cpName,:#subServiceName,:#subServiceId,:#shortCode,:#message,:#callBackUrl,:#callBackResp,:#clientIp,:#seSysIp,:#language,:#cpCallbackUrlHitTime,:#action,:#alert,:#notificationUrl,:#notificationResponse)

4. SQL批处理路由定义如下:

   <pipeline>
      <log message="Going to insert in database"></log>
      <transform>
         <method ref="insertionBean" method="subsBatchInsertion"></method>
      </transform>
      <choice>
          <when>
              <simple>${in.header.subsCount} == ${properties:batch.size}</simple>
              <to uri="sql:{{sql.subs.insertCdr}}?batch=true"></to>
              <log message="Inserted rows ${body}"></log>
          </when>
      </choice>
   </pipeline>
   

   Run Code Online (Sandbox Code Playgroud)

5. 下面是我的java代码:

   public List<Map<String, Object>> subsBatchInsertion(Exchange exchange) {
   if (subsBatchCounter > batchSize) {
       subsPayLoad.clear();
       subsBatchCounter = 1;
   }
   subsPayLoad.add(generateInsert(exchange.getIn().getBody(SubscriptionCdr.class)));
   exchange.getIn().setHeader("subsCount", subsBatchCounter);
   subsBatchCounter++;
   return subsPayLoad;
   }
   
   public Map<String, Object> generateInsert(Cdr cdr) {
   Map<String, Object> insert = new HashMap<String, Object>();
   try {
       insert …

   Run Code Online (Sandbox Code Playgroud)

我在Apache flink中有一个小用例,即批处理系统.我需要处理一组文件.每个文件的处理必须由一台机器处理.我有以下代码.始终只占用一个任务槽,并且一个接一个地处理文件.我有6个节点(所以6个任务管理器),并在每个节点配置4个任务槽.所以,我希望一次处理24个文件.

class MyMapPartitionFunction extends RichMapPartitionFunction[java.io.File, Int] {
  override def mapPartition(
      myfiles: java.lang.Iterable[java.io.File],
      out:org.apache.flink.util.Collector[Int])
    : Unit  =  {
    var temp = myfiles.iterator()
    while(temp.hasNext()){
      val fp1 = getRuntimeContext.getDistributedCache.getFile("hadoopRun.sh")
      val file = new File(temp.next().toURI)
      Process(
        "/bin/bash ./run.sh  " + argumentsList(3)+ "/" + file.getName + " " + argumentsList(7) + "/" + file.getName + ".csv",
        new File(fp1.getAbsoluteFile.getParent))
        .lines
        .foreach{println}
      out.collect(1)
    }
  }
}

我启动了flink as ./bin/start-cluster.sh命令,Web用户界面显示它有6个任务管理器,24个任务槽.

这些文件夹包含大约49个文件.当我在这个集合上创建mapPartition时,我希望跨越49个并行进程.但是,在我的基础设施中,它们都是一个接一个地处理的.这意味着只有一台机器(一个任务管理器)处理所有49个文件名.我想要的是,每个插槽配置2个任务,我希望同时处理24个文件.

任何指针肯定会有所帮助.我在flink-conf.yaml文件中有这些参数

jobmanager.heap.mb: 2048
taskmanager.heap.mb: 1024
taskmanager.numberOfTaskSlots: 4
taskmanager.memory.preallocate: false
parallelism.default: 24

提前致谢.谁能让我知道我哪里出错了？