注：

将 OCI 数据流与 Apache Spark 流处理结合使用，以可扩展且近乎实时的应用处理 Kafka 主题

简介

Oracle Cloud Infrastructure (OCI) 数据流是面向名为 Apache Spark 的开源项目的托管服务。基本上，通过 Spark，您可以将其用于处理大量文件、流处理和数据库操作。您可以使用高度可扩展的处理来构建应用程序。Spark 可以扩展和使用群集计算机来同时处理具有最低配置的任务。

通过将 Spark 用作托管服务（数据流），您可以添加许多可伸缩服务以乘以云处理的力量。数据流能够处理 Spark Streaming 。

流处理应用需要长时间的持续执行，并且通常会超过 24 小时，并且可能长达数周甚至数月。如果出现意外故障，流处理应用程序必须从故障点重新启动，而不会产生不正确的计算结果。数据流依赖 Spark 结构化流检查点来记录处理的偏移，可将其存储在对象存储桶中。

注：如果您需要将数据作为批处理策略进行处理，可以阅读以下文章：使用 Oracle Cloud Infrastructure Data Flow 在 Autonomous Database 和 Kafka 中处理大型文件

dataflow-use-case.png

在本教程中，您可以看到用于处理数据卷流、查询数据库以及合并/联接数据以在内存中形成另一个表或近乎实时地将数据发送到任何目标的最常见活动。您可以将此海量数据写入数据库，并在 Kafka 队列中写入具有非常低成本的高效性能。

目标

先决条件

任务 1：创建对象存储结构

对象存储将用作默认文件存储库。您可以使用其他类型的文件资料档案库，但对象存储是一种简单且低成本的方法来处理具有性能的文件。在本教程中，两个应用程序都将从对象存储加载大型 CSV 文件，展示 Apache Spark 如何快速而智能地处理大量数据。

  1. 创建区间：区间对于组织和隔离云资源很重要。您可以按 IAM 策略隔离资源。

    • 您可以使用此链接了解和设置区间的策略：管理区间

    • 创建一个区间来托管本教程中 2 个应用程序的所有资源。创建名为 analytics 的区间。

    • 转到 Oracle Cloud 主菜单并搜索：身份与安全性、区间。在“区间”部分中，单击创建区间并输入名称。 create-compartment.png

      注：您需要向一组用户授予访问权限并包括您的用户。

    • 单击创建区间以包括您的区间。

  2. 在对象存储中创建存储桶：存储桶是用于存储对象的逻辑容器，因此用于此演示的所有文件都将存储在此存储桶中。

    • 转到 Oracle Cloud 主菜单并搜索存储和存储桶。在“Buckets（存储桶）”部分，选择之前创建的区间 (analytics)。

      select-compartment.png

    • 单击创建存储桶。创建 4 个存储桶：应用、数据、数据流日志、Wallet

      create-bucket.png

    • 输入包含这 4 个存储桶的存储桶名称信息，并使用默认选择维护其他参数。

    • 对于每个存储桶，单击创建。您可以看到创建的存储桶。

      存储桶 -dataflow.png

注：查看存储桶的 IAM 策略。如果要在演示应用程序中使用这些存储桶，则必须设置策略。您可以在对象存储概览和 IAM 策略中了解这些概念和设置。

任务 2：创建 Autonomous Database

Oracle Cloud Autonomous Database 是 Oracle Database 的托管服务。对于本教程，应用程序将出于安全原因通过 Wallet 连接到数据库。

注：在此处查看用于访问 Autonomous Database 的 IAM 策略：IAM Policy for Autonomous Database

任务 3：上载 CSV 示例文件

为了演示 Apache Spark 的强大功能，应用程序将读取包含 1,000,000 行的 CSV 文件。此数据将仅使用一个命令行插入 Autonomous Data Warehouse 数据库，并在 Kafka 流处理 (Oracle Cloud Streaming) 上发布。所有这些资源都可扩展且非常适合高数据量。

您可以看到名为 GDPPERCAPTA 的新表已成功导入。

adw-table-imported.png

任务 4：为 ADW ADMIN 密码创建密钥储存库

出于安全原因，ADW ADMIN 密码将保存在 Vault 上。Oracle Cloud Vault 可以安全地托管此密码，并且可以通过 OCI 验证在您的应用中访问。

注：请在此处查看 OCI Vault 的 IAM 策略： OCI Vault IAM 策略。

任务 5：创建 Kafka 流处理 (Oracle Cloud Streaming)

Oracle Cloud Streaming 是一种类似于托管流处理服务的 Kafka。您可以使用 Kafka API 和常用 SDK 开发应用。在本教程中，您将创建流处理实例并配置为在两个应用程序中执行以发布和使用大量数据。

  1. 从 Oracle Cloud 主菜单中，转至 Analytics & AI 、 Streams 。

  2. 将区间更改为 analytics 。将在此区间中创建此演示中的每个资源。这样可以更安全、更轻松地控制 IAM。

  3. 单击创建流。

    create-stream.png

  4. 将名称输入为 kafka_like （例如），您可以使用默认值维护所有其他参数。

    save-create-stream.png

  5. 单击创建以初始化实例。

  6. 等待 Active（活动）状态。现在，您可以使用该实例。

    注：在流创建过程中，您可以选择自动创建默认流池选项以自动创建默认池。

  7. 单击 DefaultPool 链接。

    default-pool-option.png

  8. 查看连接设置。

    stream-conn-settings.png

    kafka-conn.png

  9. 在下一步中根据需要添加此信息注释。

注：请在此处查看 OCI 流处理的 IAM 策略： OCI 流处理的 IAM 策略。

任务 6：生成 AUTH TOKEN 以访问 Kafka

您可以使用与 OCI IAM 上的用户关联的验证令牌来访问 Oracle Cloud 中的 OCI 流处理 (Kafka API) 和其他资源。在 Kafka 连接设置中，SASL 连接字符串有一个名为 password 的参数和一个 AUTH_TOKEN 值，如上一个任务中所述。要启用对 OCI 流处理的访问，您需要转到 OCI 控制台上的用户并创建 AUTH TOKEN。

  1. 从 Oracle Cloud 主菜单中，转至身份与安全性、用户。

    注：请记住，创建 AUTH TOKEN 所需的用户是配置有 OCI CLI 的用户，以及直到现在为止创建的资源的所有 IAM 策略配置。资源为：

    • Oracle Cloud Autonomous Data Warehouse
    • Oracle Cloud 流处理
    • Oracle Object Storage
    • Oracle Data Flow

  2. 单击您的用户名以查看详细信息。

    auth_token_create.png

  3. 单击控制台左侧的 Auth Tokens 选项，然后单击 Generate Token（生成令牌）。

    注：标记将仅在此步骤中生成，在完成该步骤后将不可见。因此，请复制值并保存该值。如果丢失令牌值，则必须再次生成验证令牌。

auth_token_1.png auth_token_2.png

任务 7：设置演示应用程序

本教程有一个演示应用程序，我们将为其设置所需的信息。

  1. 使用以下链接下载应用程序：

  2. 在 Oracle Cloud 控制台中找到以下详细信息：

    • 租户名称空间

      租户 -namespace-1.png

      租户名称空间 -detail.png

    • 密码密钥

      Vault-adw.png

      vault-adw-detail.png

      机密 -adw.png

    • 流处理连接设置

      kafka-conn.png

    • 验证标记

      auth_token_create.png

      auth_token_2.png

  3. 打开下载的 zip 文件（Java-CSV-DB.zip 和 JavaConsumeKafka.zip）。转到 /src/main/java/example 文件夹并查找 Example.java 代码。

    代码 -variables.png

    这些变量需要随租户资源值一起更改。

    变量名称 资源名称 信息标题
    bootstrapServers 流处理连接设置 引导服务器
    streamPoolId 流处理连接设置 ocid1.streampool.oc1.iad..... SASL 连接字符串中的值
    kafkaUsername 流处理连接设置 SASL 连接字符串中 "" 内的用户名的值
    kafkaPassword 验证标记 该值仅显示在创建步骤中
    OBJECT_STORAGE_NAMESPACE 租户名称空间 租户
    名称空间 租户名称空间 租户
    PASSWORD_SECRET_OCID PASSWORD_SECRET_OCID OCID

注：为此演示创建的所有资源都位于 US-ASHBURN-1 区域中。检查要使用的区域。如果更改区域，则需要更改 2 个代码文件中的 2 个点：

任务 8：了解 Java 代码

此教程是用 Java 创建的，此代码也可以移植到 Python。为了证明效率和可扩展性，开发了应用程序以在集成过程的常见用例中显示一些可能性。因此，应用程序的代码显示以下示例：

此演示可以在本地计算机中执行，并部署到数据流实例中，以作为作业执行运行。

注：对于数据流作业和本地计算机，请使用 OCI CLI 配置访问 OCI 资源。在“数据流”端，所有内容都已预先配置，因此无需更改参数。在本地计算机端，您应该已安装 OCI CLI 并配置租户、用户和私有密钥来访问您的 OCI 资源。

让我们在以下部分中显示 Example.java 代码：

任务 9：使用 Maven 为您的应用打包

在 Spark 中执行作业之前，必须使用 Maven 打包应用程序。

  1. 转到 /DataflowSparkStreamDemo 文件夹并执行以下命令：

    mvn package

  2. 您可以看到 Maven 正在开始打包。

    maven-package-1a.png

  3. 如果一切正确，您可以看到成功消息。

    maven-success-1a.png

任务 10：验证执行

  1. 通过运行以下命令在本地 Spark 计算机中测试应用程序：

    spark-submit --class example.Example target/consumekafka-1.0-SNAPSHOT.jar

  2. 转到 Oracle Cloud Streaming Kafka 实例，然后单击生成测试消息以生成一些数据来测试实时应用程序。

    测试 -kafka-1.png

  3. 您可以将此 JSON 消息放入 Kafka 主题。

    {"Organization Id": "1235", "Name": "Teste", "Country": "Luxembourg"}

    测试 -kafka-2.png

  4. 每次单击生成时，都会向应用程序发送一条消息。您可以查看应用程序的输出日志，如下所示：

    • 这是从 kafka 主题读取的数据。

      测试 -output-1.png

    • 这是 ADW 表中的合并数据。

      测试 -output-2.png

任务 11：创建和执行数据流作业

现在，在本地 Spark 计算机成功运行两个应用后，您可以将其部署到租户中的 Oracle Cloud 数据流中。

注：要配置对 Oracle 对象存储和 Oracle 流处理 (Kafka) 等资源的访问，请参阅 Spark 流处理文档：启用对数据流的访问

  1. 将程序包上载到对象存储中。

    • 在创建数据流应用程序之前，您需要将 Java Artifact 应用程序（您的 ***-SNAPSHOT.jar 文件）上载到名为 apps 的对象存储存储桶中。

  2. 创建数据流应用程序。

    • 选择 Oracle Cloud 主菜单，然后转至分析和 AI 和数据流。在创建数据流应用程序之前，请确保选择分析区间。

    • 单击创建应用程序。

      create-dataflow-app.png

    • 填充此类参数。

      数据流 -app.png

    • 单击创建。

    • 创建后，单击缩放演示链接可查看详细信息。要运行作业，请单击 RUN 。

      注：单击显示高级选项可为 Spark Stream 执行类型启用 OCI 安全性。

      高级 -options.png

  3. 激活以下选项。

    主体 -execution.png

  4. 单击运行以执行作业。

  5. 确认参数，然后再次单击运行。

    dataflow-run-job.png

    • 可以查看作业的状态。

      dataflow-run-status.png

    • 等待状态转至成功，您可以看到结果。

      dataflow-run-success.png

确认

更多学习资源

探索 docs.oracle.com/learn 上的其他实验室，或者访问 Oracle Learning YouTube 频道上的更多免费学习内容。此外，请访问 education.oracle.com/learning-explorer 成为 Oracle Learning Explorer。

有关产品文档，请访问 Oracle 帮助中心。