标签: dataflow

我有一个我正在努力的任务,让我难过.希望你能帮助我.我正在使用数据流任务,它基本上是在sqlite表中插入一行.我使用"SQL任务"这样做,但不幸的是,成功地将一个guid插入sqlite表的唯一方法是使用数据流任务将其转换为字节流.我不想使用源数据库,因为我的数据没有从一个表流到另一个表.我真的只想获取我填充的变量并将它们转换为字节流,然后我可以成功地插入到sqlite数据库中.问题是,我不能在没有源数据库的情况下使用数据流任务.

到目前为止,我的解决方法是声明一个源数据库/表,只有一列(但从不在数据流中使用它).这工作正常,我无法使用我的预设变量将行插入sqlite,但每次执行此操作时,我的输出日志中都会留下一些有点烦人的消息:

警告:0x80047076 at,SSIS.Pipeline:输出"OLE DB源输出"(11)上的输出列""(117)和组件"OLE DB源"(1)随后未在数据流任务中使用.删除此未使用的输出列可以提高数据流任务性能.

任何人都知道一个很好的方法来得到这个警告不显示？

我有一个平面文件源,它从平面文件中读取数据.我们最近在这个平面文件中添加了一个新列.

平面文件数据被插入数据库表中.为了适应目标组件中的新字段,我使用该ALTER TABLE语句将新列添加到表中.这是我所做的唯一改变.

平面文件和目标组件之间的映射是否应自动更改？我没有在SSIS包中的任何位置看到平面文件中存在其他列.

如何在SSIS包中的平面文件中配置附加列,以便平面文件源可以将数据传递到目标组件？

我是C#TPL和DataFlow的新手,我正在努力研究如何实现TPL DataFlow TransformManyBlock.我正在将其他一些代码翻译成DataFlow.我的(简化)原始代码是这样的:

private IEnumerable<byte[]> ExtractFromByteStream(Byte[] byteStream)
{
    yield return byteStream; // Plus operations on the array
}

在另一种方法中,我会这样称呼它:

foreach (byte[] myByteArray in ExtractFromByteStream(byteStream))
{
    // Do stuff with myByteArray
}

我正在尝试创建一个TransformManyBlock来生成来自较大输入数组(实际上是二进制流)的多个小数组(实际上是数据包),因此in和out都是类型byte[].

我尝试了下面的内容,但我知道我错了.我想使用与以前相同的函数构造这个块,并且只是将TransformManyBlock它包裹起来.我收到一个错误"这个电话很模糊......"

var streamTransformManyBlock = new TransformManyBlock<byte[], byte[]>(ExtractFromByteStream);

我是Apache Nifi的新手,我有一个用例,我需要解析和解码来自传感器的不同类型的消息,转换并加载Hbase中的数据我的所有传感器每隔10分钟通过API通过一个请求发送数据,我现在所做的是一个JAVA服务,它监听特定端口并执行所有ETL数据流,任何想法如何使用Apache Nifi用于此用例？

我在varchar列'A'上有一个非Clustered索引的表.
当我使用Order By A子句时,我可以看到它扫描索引并在几秒钟内给出结果.
但是当我使用SSIS的Sort Component作为列'A'时,我可以看到对记录进行排序需要几分钟.

所以我理解它无法识别我的非聚集索引

有没有人有任何想法使用SSIS索引但不使用查询而不是组件？

抱歉，如果已经有类似的问题，我找不到。

我有以下情况：

1. 我必须对图像进行一些处理，并且 TPL 数据流非常适合这里，因为它允许我轻松地并行执行工作流的不同部分，并以逻辑单元分隔代码
2. 有一个我无法控制的函数返回图像列表。它用作我的网格（或管道，更准确地说）中的第二个节点
3. 我在 Dataflow 网格中的所有其他节点都使用单个图像，因此我在第二个项目符号中提到的 I 之后的节点期望获得一个图像（这对于并行性很重要）

是否有一个块（或其他一些解决方案）我可以使用它会接受类型IEnumerable<T>或类似的输入并将其每个元素转发IEnumerable到一个期望接收的块T？

我不想重新发明轮子，所以我想在深入研究 API 并尝试编写自定义块之前检查是否有简单的解决方案。此外，将错误和完成传播到管道末端也很重要。

谢谢你的回答！

没有关于如何将pCollections转换为输入到.CoGroupByKey()所需的pCollections的文档

上下文基本上我有两个大的pCollections,我需要能够找到两者之间的差异,对于第二类ETL更改(如果它在pColl1中不存在,那么添加到pColl2中的嵌套字段),这样我就能够从BigQuery保留这些记录的历史记录.

管道架构:

1. 将BQ表读入2个pCollections:dwsku和产品.
2. 将CoGroupByKey()应用于两个集合以返回 - >结果
3. 解析结果以查找并将dwsku中的所有更改嵌套到产品中.

建议任何帮助.我在SO上发现了一个java链接,它做了我需要完成的同样的事情(但是在Python SDK上没有任何内容).

从PCollection <TableRow>转换为PCollection <KV <K,V >>

是否有Apache Beam的文档/支持,特别是Python SDK？

我正在尝试完成以下操作： Beam / Dataflow中的批处理PCollection

上面链接中的答案是Java，而我使用的语言是Python。因此，我需要一些帮助来获得类似的构造。

具体来说我有这个：

 p = beam.Pipeline (options = pipeline_options)
 lines = p | 'File reading' >> ReadFromText (known_args.input)

此后，由于我的用例需要一组行，因此我需要创建另一个，PCollection但要包含ListN行的“行”。我无法逐行操作。

我尝试了一个ParDo函数，该函数使用变量将计数与N行计数器相关联，并在groupBy使用之后Map。但是这些每1000条记录会重置一次，所以这不是我想要的解决方案。我阅读了链接中的示例，但我不知道如何在Python中执行类似的操作。

我尝试将计数器保存在数据存储区中，但是，使用数据存储区读写数据流之间的速度差异非常明显。

正确的方法是什么？我不知道该怎么办。问候。

我想使用ggplot2和修改现有的桑基图ggalluvial，使其更具吸引力

我的例子来自https://corybrunson.github.io/ggalluvial/articles/ggalluvial.html

library(ggplot2)
library(ggalluvial)

data(vaccinations)
levels(vaccinations$response) <- rev(levels(vaccinations$response))
ggplot(vaccinations,
       aes(x = survey, stratum = response, alluvium = subject,
           y = freq,
           fill = response, label = response)) +
  scale_x_discrete(expand = c(.1, .1)) +
  geom_flow() +
  geom_stratum(alpha = .5) +
  geom_text(stat = "stratum", size = 3) +
  theme(legend.position = "none") +
  ggtitle("vaccination survey responses at three points in time")

^{由reprex 包于 2020 年 10 月 1 日创建(v0.3.0)}

现在，我想更改此图，它看起来类似于https://sciolisticramblings.wordpress.com/2018/11/23/sankey-charts-the-new-pie-chart/中的图，即 1. 更改绝对值到相对值（百分比） 2. 添加百分比标签并 3. …

我正在尝试重现本教程以在数据流上运行 Flex 模板。

当我提交作业时，我可以在控制台中看到它，但它尚未启动并标记为Queued。这是否意味着作业以FlexRS 模式提交？提交作业后如何直接开始作业？