问题标签 [dataflowtask]

我有一个包含不同 SQL 文件的文件夹。现在在 SSIS 包中，在 foreach 循环中，我正在读取存储在每个 sql 文件中的查询（带有脚本任务），然后在执行脚本任务中执行查询，从而在变量中获得结果集。

现在我想生成与这些结果集变量中的每一个相对应的平面文件。

这个过程是否可行（我曾尝试使用 dataflowtask，但它没有使用变量作为源）还是我应该采用不同的方法？

我创建了一个 ssis 包。在数据流任务中，我将数据从 oledbsource 传递到平面文件目标。我想在平面文件中保留空值，但它是空白的。

需要一些帮助。查看 Gpars 数据流/管道，但有些我不明白

如果您查看下面的示例（我已经使用运算符、管道线、chainWith 完成了此操作并遇到了同样的问题）。

在此示例中，我使用了任务，但也很容易没有任务，并且出现相同的问题。在此示例中，我设置了两个 DataflowQueue，一个用于初始条件，一个用于根据谓词评估的结果。然后我布局一个管道，根据谓词（偶数测试）评估输入并将结果存储在输出结果队列中

设置了管道并将一些条目发布到第一个队列中，我相信这些条目将在数据可用时被处理（这也不适用于操作员版本），如您所见，我测试了 resultQ 的大小为零（如果我在我将条目写入 sessionQ 之后，删除仍然正确的任务）。因此，写入数据不会“触发”处理。

第一个任务将一些条目保存到队列中。

它仅在第二个任务/chainWith 等中 - 您在引擎开始运行的第二个队列上调用 .val （或 get() ），所有条目都从第一个队列处理，结果绑定到 resultQ。

您可以从断言中看到这一点，因为一旦进行了第一个触发器 (.val) 同步调用，引擎就会运行并处理起始 sessionQ 中的所有绑定条目。

这是一个问题，因为直到您运行第一个 .val 调用 - 如果您执行 poll() 或 resultQ.interator.size() 例如它是空的且未绑定，size()=0。所以你不能写

因为它总是空的，直到您使用 sessionQ 中的第一个项目。我不明白为什么？在条目绑定到第一个 dataflowQueue 后，我认为这些项目将在它们可用时被消耗（已绑定） - 但它们不是。

现在这很棘手，因为您无法通过，检查结果的大小，在 resultQ 上执行 poll()，因为它会失败，直到读取 sessionQ 的第一个 DF。

我最终不得不使用初始值数组的大小（告诉我保存到队列中的条目），因为唯一意味着从 resultQ 中读取相同的数字以清空它（在上面我只消耗了 3 resultsQ 中的记录和断言表明 resultQ 中还剩下 2 条记录（但只有在第一次 .val 调用之后，如果您注释掉所有断言开始失败）

我用 Dataflow.operator、Pipeline 等尝试了这个并得到了同样的问题。为什么每个输入都绑定到 SessionQ 时，工作没有得到处理？

最后，在 Pipeline 的情况下，有一个 .complete() 方法，如果你在管道中处理闭包 {}，它会保持打开状态 (!complete())，但是当你运行像 .binaryChoice() 这样的方法时，它会标记管道已完成，无法添加进一步的操作。为什么这样做？

当然，我不明白那个状态在说什么（不再进行处理），如果您尝试在这种方法之后再执行另一个步骤，则会引发异常。

无论哪种方式 - 我尝试过这样的管道线

但是，当您将值绑定到 Q 时，什么都不会发生 - 直到您使用类似的输出

当管道突然“运行”并处理存储在 Q 中的所有 DF 项目时。

除了第一个 .val 消耗之外，我没有尝试过启动工作的安排

任何人都可以解释为什么会这样，我必须在这里忽略这一点，但是在读取第一个条目之前“什么都不做”不是我所期望的，并且会使任何大小评估（.iterator.size（），poll（）等无效) 类型调用 DataflowWriteChannel 目标。

我很感激对此的任何帮助-我已经为此苦苦挣扎了两天，但一无所获。我也查看了所有 Gpars 测试，它们只是调用 .val 的次数与输入绑定的次数相同 - 所以不要显示我描述的问题。

Vaclav Pech 或任何其他观看这些问题的 Gpars 大师，我将不胜感激任何有关这方面的帮助见解，以帮助我度过这个难关

提前问候

我有一个简单的 DTSX，它接收一些数据，并将其写入 SQL 服务器数据库，在此 DTSX 执行中，它在某些字段上引发了一些错误：

但正如您从 OLEDB DESTINATION COLUMNS MMAPING 中看到的那样，这不是真的……怎么会？

我有一个 ssis 包，它从 excel 文件中导出数据并将其转储到 SQL 表中。对于处理文件，我使用foreach循环和数据流打开一个 excel 源并将数据转储到 ole db 目标。如果任何文件不包含所需的选项卡，我希望 ssis 包记录错误并移至下一次迭代。我尝试了以下操作，但包失败：

1. 传播 = 假
2. ForceExecutionResult = 成功

我该如何处理？

附图是控制流、数据流和进度的屏幕截图。

这是场景：

第一个任务检查某些文件的存在，写入变量并使用表达式检查运行哪个数据流任务。在这里一切正常。

我希望在完成之前的所有任务后打开最后一个任务。有些任务可能无法执行。也许其中只有四个已被处理。表达式中的布尔变量决定了它的执行。

我尝试将约束设置为“表达式或约束”并检查结果是否为正，并且所有约束都按逻辑 OR 分组。但是，当其中一个任务结束时，会启动最后一个任务。相反，我应该等待所有任务完成后再开始最后一个任务。

如果设置了不同的 cotraints 分组（逻辑 AND），则最后一个任务不会运行，因为某些数据流任务没有执行。

有任何想法吗？

谢谢。

我正在尝试找到一种方法来重新排序我的 Kafka 消息并将有序消息发送到使用Apache Beam和Google DataFlow的新主题。

我有发送以下格式的字符串消息的 Kafka 发布者： {system_timestamp}-{event_name}?{parameters}

例如：

我想做的是根据消息的{system-timestamp}部分并在 5 秒的窗口内重新排序事件，因为我们的发布者不保证消息将按照{system-timestamp}值发送。

我编写了一个模拟排序器函数，对从 Kafka 接收到的事件进行排序（使用KafkaIO源）：

因此，我使用GroupByKey.<String, String>create()转换对消息进行了分组，在 sortrin 事件之后，我需要以某种方式将它们从KafkaIO值转换KV<String, Iterable<String>>为接受。KV<String, String> or KV<Void, String>所以我想做的就是忽略通过分组转换键创建的，只需将每个值作为单独的消息传递给 KafkaIO writer。

我探索了Combine#perKey转换，但它接受只能将所有值组合到一个字符串（带有一些分隔符）的SerializableFunctionKafkaIO#read() ，因此我只将一个值作为一个连接字符串而不是每个值（由 读取）传递给 KafkaIO 写入器。

如何以编程方式为 DataFlow 设置 AutoAdjustBufferSize 属性？IE

我应该从源中读取数据，其列名永久更改（列名是日期）：

我想把它写在一个表中，我有这样一个映射：

当然，如果列名被更改，到下一次运行时，它就不再起作用了。你有什么想法，我怎么能动态指定 coumn 的名称？或者我怎样才能在广播中阅读栏目名称？

是否可以通过每次运行生成具有新列名的新表？我怎样才能做到这一点？

我有一种情况，在源表中，有 4 列具有数据类型浮点数。当我创建具有相同架构的新目标表（仅具有数据类型浮点）时，值加载为 0。当我手动运行 SSIS 传输数据时不会发生这种情况，但当从 SQL 代理运行 ssis 包时不会发生这种情况。有没有这样的机会将浮点值转换为 0？