我正在尝试使用 cdap 实时管道完成 pubsub 订阅。

我可以连接 pubsub，但属性列作为 MAP 数据类型通过，我看到无法对它做任何事情（我需要其中的数据）。

这个想法是获取该消息并将其放入数据库中以进行进一步处理。

有什么方法可以获取 MAP 数据类型并将其转换为有用的东西？

我正在尝试在自定义操作插件中获取数据沿袭元数据，例如数据源/模式和数据目标/模式，该插件在管道中的其他步骤成功运行后执行。

我有一个可以执行的基本 Action 插件，但我无法找到获取所需元数据的方法。

我正在处理的用例是将数据沿袭推入第三方数据治理工具。

如果有人能指出我正确的方向，我将不胜感激！

我有一个 GCP 数据融合管道，我在其中对一个 API 执行 GET 请求，该请求返回一个包含用户 ID 在内的用户信息的 JSON 列表。我可以使用 Data Fusion HTTP 插件（在 Data Fusion HUB 中提供）成功地做到这一点。以下是此类列表的示例：

基于这个列表（我已经成功解析），我想在另一个 API 上基于每个用户 ID 进行 HTTP 调用，而不是仅仅将它下沉到数据库中。有没有办法在单个数据融合管道上执行此操作（无需启动另一个管道来执行一个 HTTP 请求？

我尝试使用 Data Fusion python 转换插件（在 HUB 中可用）并使用 pythonrequests库来执行 http 请求，但这最终导致错误，因为运行转换插件的 python 解释器中没有安装请求库。

我们有一个由 Cloud Composer DAG 触发的 DataFusion 管道。此管道提供一个临时 DataProc 集群，该集群在理想情况下会在完成任务后终止。

在我们的案例中，有时，并非总是如此，这个短暂的 DataProc 集群会卡在运行状态。集群内部的作业也处于运行状态，最后的日志信息如下：

在 DataFusion 端，管道标记为成功。DataFusion 日志如下：

任何想法是什么导致了这个问题？

ps：消息中的标识符被替换为随机值

我正在尝试为 GCP Data Fusion 开发自己的插件。所以我按照文档，并从https://github.com/data-integrations/example-transform克隆了示例。

但是在构建项目时，我遇到了导入测试所需的依赖项的问题：

我收到以下错误：

我能做些什么来解决这个问题并测试我的插件？

我正在尝试在 Google Cloud Data Fusion 中自动化一些管道执行（我们目前使用的是 6.1.4 和 6.4.0）。此时，我们正在通过 PUT API 调用将一些“运行时参数”注入 DF。我的问题是关于注入参数来修改配置部分。例如，我们目前正在使用“ system.profile.name ”参数来告诉此管道使用特定的配置文件，请参阅： 运行时参数的屏幕截图。

我想知道是否有任何类似的配置选项来定义“配置/资源/执行器内存”标签：“配置/资源”标签的屏幕截图。我知道这可以通过修改 UI 或在导入管道之前在管道模板 (json) 中设置不同的值来手动配置。但是我想知道在部署管道后是否有任何方法可以自动执行此操作（我不想每次修改时都重新部署管道）。

提前致谢！

我正在从 GCS 读取文本/csv 文件，并希望生成一个 excel 文件并将其发布回 GCS。下面是我正在使用的 Spark 代码，它正在创建一个 excel 文件，但没有数据并且已损坏。管道引发以下错误。我在依赖项中包含了 xmlbeans-3.1.0.jar JAR 文件的路径，但看起来它没有从指定位置获取最新版本。

如何覆盖类路径以使用 CDAP/Data Fusion 中指定位置的 JAR 文件？

管道

我正在使用的火花代码是：

谢谢， NS

我正在尝试在 Datafusion 上创建一个管道以从 bigquery 中读取具有STRUCT类型的表，但收到此错误：

根据文档，可以选择使用 6.2 及更高版本中的现有 Dataproc 集群。

我们使用 Cloud Data Fusion 6.2.0 ，但当我们尝试创建新的计算配置文件时 ，现有的 Dataproc不会出现。

我们做错了什么？为什么所描述的选项不显示？我们需要做一些额外的配置吗？

更新 1

当我选择 Dataproc 时，我看到以下内容：

更新 2

当我们尝试使用Remote Hadoop Provisioner时，我们在 /logs/program.log 文件中收到以下错误消息。SSH 连接成功，因为 run-id 文件夹在那里。

我在大数据领域以及 CDAP 环境中都是非常陌生的。我目前正在完成关于电池分析数据管道的最后论文，其中还包括其分析。我正在使用 CDAP，因为我的主管说这很容易。如果我的一些问题在 CDAP 的文档中可用，我很抱歉。

以下是我提出和实施的数据管道（请仔细检查，因为它会被进一步询问）：

下面是它在 CDAP 上的实现：

*我每分钟使用 cron，最大并发数为 5。

到目前为止，唯一缺少的是 Python 程序。

但是，当我部署此管道时，出现了一个特殊问题：

• 它有时会以“必要的服务遇到间歇性问题”的消息停止，然后是自我修复协议。然而，在我的记录中没有表明自我修复确实有效。所以，我必须在 CLI 上重新启动我的 CDAP。

上述错误主要是由于我的计算机上有活动，尤其是在 CDAP 的环境中。我的问题是，是否有任何解决方案可以防止这些错误。也许在 CLI 上部署数据管道？

我想问的另一个问题是，有没有办法在 CDAP 的环境中部署 Python 程序（ANN 程序）？据我所知，CDAP 上只有一个 Python 插件可用，这是我认为不适合我用例的转换插件。

至于现在，我必须同时运行 Python 程序和 CDAP 的使用 Cron 程序，我相信这种方法容易失败和错误。

TLDR；

1. 我的系统一直卡在“必要的服务遇到间歇性问题”消息并且无法自我修复并且必须手动重新启动。有什么解决办法吗？
2. 是否可以通过 CLI 运行 CDAP？
3. 是否可以使用嵌入在 CDAP 管道中的 ANN 来部署 Python 代码来预测从管道下载的数据？

PS：我标记了 Cloud Data Fusion，因为它基于 CDAP 的基础架构

这就是我想问的。谢谢你。