Spring Cloud Data Flow初探

Posted 2023-04-19

参考技术A 先看一下官方给出的定义

SCDF (Spring Cloud Data Flow)的核心功能是ETL （Extract, Transform, Load ），Extract，Transform，Load 分别对应上图的Source，Processor 和Sink，这三个组件是Spring Boot 微服务，部署运行在SCDF之上的，三个微服务放在一起构成一个Stream（pipeline）用来实现数据处理，它们之间通过AMQP进行异步的消息传递。

SCDF 使用了 Spring Cloud stream 模块。我们Stream创建和运行以 Spring Boot 应用为形式的消息传递微服务，以便它们可以部署在不同的平台上，独立运行并相互交互。在使用 Spring Cloud stream 模块创建数据管道时，SCDF 可以充当类似胶水的角色，旨在于提供一个管理服务模型，目的是用于精简数据项目的工程量，并让开发人员将精力集中在具体问题及对问题的分析上

如上图所示，开发人员可以通过编写JavaDSL或者用拖拽的方式完成一个Pipeline的定义，定义好的Pipeline（Stream）可以一键部署到SCDF Server，一个数据流处理的管道就运行起来了，开发人员只需要关注Extract, Transform, Load （Source，Processor，Sink）三个微服务的业务逻辑开发，因为它们都是基于Spring Boot的项目，所以开发起来也非常简单，可以说SCDF真的让我们只于去解决问题，其它的事情都可以交给SCDF解决。当然SCDF很多默认的微服务组件可以用，很多时候我们不需要开发了，只需要找到合适的application （微服务）用于组成我们想要的Stream就可以了。

SCDF 由下面的Spring Cloud家族成员组成

参考
https://www.javainuse.com/spring/cloud-data-flow
https://www.e4developer.com/2018/02/18/getting-started-with-spring-cloud-data-flow/
https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle

Spring Cloud Data Flow 的自定义任务中缺少参数

【中文标题】Spring Cloud Data Flow 的自定义任务中缺少参数

【英文标题】：Missing parameters in custom Task for Spring Cloud Data Flow

【发布时间】：2020-10-15 00:02:07

【问题描述】：

我正在开发一个使用 Spring Cloud Data Flow 的任务，但是我看不到我的任务（这是一个 Spring Boot 应用程序）的执行参数或如何执行该任务。

我将我的应用程序作为基础导入，并将项目作为基础导入：spring-cloud-starter-task-timestamp

这是我的 pom，我使用 Spring Initializr 制作，然后使用时间戳任务应用程序进行调整。

<groupId>teste</groupId>
<artifactId>scdf-neo4j-task</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>scdf-neo4j-task</name>
<description>scdf-neo4j-task</description>

<parent>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-task-parent</artifactId>
    <version>2.3.0-SNAPSHOT</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>

<properties>
    <spring-cloud.version>Hoxton.SR5</spring-cloud.version>
    <start-class>teste.scdf.Neo4jTaskApplication</start-class>
    <skipInstall>true</skipInstall>
</properties>

<dependencies>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>

    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-task</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-configuration-processor</artifactId>
        <optional>true</optional>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-jdbc</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-configuration-processor</artifactId>
        <optional>true</optional>
    </dependency>

    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
    </dependency>

    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>

    <dependency>
        <groupId>org.neo4j.driver</groupId>
        <artifactId>neo4j-java-driver</artifactId>
        <version>4.0.1</version>
    </dependency>

</dependencies>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>$spring-cloud.version</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-task-dependencies</artifactId>
            <version>2.3.0-SNAPSHOT</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
        </plugin>
    </plugins>
</build>

<repositories>
    <repository>
        <id>spring-snapshots</id>
        <name>Spring Snapshots</name>
        <url>https://repo.spring.io/libs-snapshot-local</url>
        <snapshots>
            <enabled>true</enabled>
        </snapshots>
    </repository>
    <repository>
        <id>spring-milestones</id>
        <name>Spring Milestones</name>
        <url>https://repo.spring.io/libs-milestone-local</url>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
    <repository>
        <id>spring-releases</id>
        <name>Spring Releases</name>
        <url>https://repo.spring.io/release</url>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
    <repository>
        <id>spring-libs-release</id>
        <name>Spring Libs Release</name>
        <url>https://repo.spring.io/libs-release</url>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
    <repository>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
        <id>spring-milestone-release</id>
        <name>Spring Milestone Release</name>
        <url>https://repo.spring.io/libs-milestone</url>
    </repository>
</repositories>
<pluginRepositories>
    <pluginRepository>
        <id>spring-releases></id>
        <name>Spring Releases</name>
        <url>https://repo.spring.io/libs-release</url>
    </pluginRepository>
    <pluginRepository>
        <id>spring-snapshots</id>
        <name>Spring Snapshots</name>
        <url>https://repo.spring.io/libs-snapshot-local</url>
        <snapshots>
            <enabled>true</enabled>
        </snapshots>
    </pluginRepository>
    <pluginRepository>
        <id>spring-milestones</id>
        <name>Spring Milestones</name>
        <url>https://repo.spring.io/libs-milestone-local</url>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </pluginRepository>
</pluginRepositories>
</project>

这是我实现 CommandLineRunner 的主要 Spring Boot 类：

@Log
@EnableTask
@SpringBootApplication
@EnableConfigurationProperties(Neo4jTaskProperties.class)
public class Neo4jTaskApplication


    @Autowired
    public Neo4jTaskProperties neo4jTaskProperties;

    public static void main(String[] args)
    
        SpringApplication.run(Neo4jTaskApplication.class, args);
    

    @Bean
    public CommandLineRunner commandLineRunner()
    
        return args ->
        
            Driver driver = GraphDatabase.driver(
                    neo4jTaskProperties.getUri(),
                    AuthTokens.basic(
                            neo4jTaskProperties.getUsername(),
                            neo4jTaskProperties.getPassword()
                    )
            );

            try (Session session = driver.session())
            
                session.writeTransaction(transactionWork ->
                
                    ResultSummary resultSummary = transactionWork.run(neo4jTaskProperties.getCommand()).consume();

                    return resultSummary;
                );
            
        ;
    

这是我的属性类：

@Data
@ConfigurationProperties("neo4j")
public class Neo4jTaskProperties 

    private String uri;

    private String username;

    private String password;

    private String command;

我该如何解决这个问题？

【参考方案1】：

您需要将您的任务应用程序属性添加到 Spring Cloud Data Flow 知道如何提取并使其可用的应用程序 metadata。

您可以参考这个page，它会引导您了解如何添加元数据。

此外，您可以查看此sample 配置作为如何添加此类元数据的参考。