lanyuanxiaoyao/hudi-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature(executor): 增加Flink集群任务模块

Browse Source 
新增 executor-manager 和 executor-task
manager管理任务信息启停等，task执行业务方法
This commit is contained in:
兰缘小妖
2023-12-05 14:53:15 +08:00
parent 5450559470
commit 62bfc08fc3
18 changed files with 715 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
bin/build-executor-manager.sh Executable file
View File

@@ -0,0 +1,4 @@
#!/bin/bash
mvn -pl service-dependencies,service-configuration,service-forest,service-executor,service-executor/service-executor-core clean deploy -D skipTests -P local -s ~/.m2/settings-development.xml
mvn -pl service-executor/service-executor-manager clean package -D skipTests -s ~/.m2/settings-development.xml -P b2e1
ytp-transfer2 /Users/lanyuanxiaoyao/Project/IdeaProjects/hudi-service/service-executor/service-executor-manager/target/service-executor-manager-1.0.0-SNAPSHOT.jar

4
bin/build-executor-task.sh Executable file
View File

@@ -0,0 +1,4 @@
#!/bin/bash
mvn -pl service-dependencies,service-configuration,service-forest,service-executor,service-executor/service-executor-core clean deploy -D skipTests -P local -s ~/.m2/settings-development.xml
mvn -pl service-executor/service-executor-task clean package -D skipTests -s ~/.m2/settings-development.xml
ytp-transfer2 /Users/lanyuanxiaoyao/Project/IdeaProjects/hudi-service/service-executor/service-executor-task/target/service-executor-task-1.0.0-SNAPSHOT.jar

5
pom.xml
View File

@@ -22,6 +22,7 @@
<module>service-cli</module>
<module>service-loki-query</module>
<module>service-test-query</module>
<module>service-executor</module>
</modules>
<properties>
@@ -33,6 +34,10 @@
<spring-boot.version>2.6.8</spring-boot.version>
<spring-cloud.version>2021.0.3</spring-cloud.version>
<flink.major.version>1.13</flink.major.version>
<scala.major.version>2.11</scala.major.version>
<hudi.version>0.12.0-eshore-SNAPSHOT</hudi.version>
<flink.version>1.13.3</flink.version>
</properties>
<profiles>

3
service-cli/service-cli-runner/src/main/resources/application.yml
View File

@@ -113,3 +113,6 @@ deploy:
- name: service-flink-query
source-jar: service-flink-query-1.0.0-SNAPSHOT.jar
replicas: 4
- name: service-executor-manager
source-jar: service-executor-manager-1.0.0-SNAPSHOT.jar
replicas: 1

26
service-configuration/src/test/java/com/lanyuanxiaoyao/service/configuration/TestSecurityDecrypt.java Normal file
View File

@@ -0,0 +1,26 @@
package com.lanyuanxiaoyao.service.configuration;
import org.jasypt.encryption.pbe.PooledPBEStringEncryptor;
import org.jasypt.encryption.pbe.config.SimpleStringPBEConfig;
/**
* @author lanyuanxiaoyao
* @date 2023-12-05
*/
public class TestSecurityDecrypt {
public static void main(String[] args) {
PooledPBEStringEncryptor encryptor = new PooledPBEStringEncryptor();
SimpleStringPBEConfig config = new SimpleStringPBEConfig();
config.setPassword("r#(R,P\"Dp^A47>WSn:Wn].gs/+\"v:q_Q*An~zF*g-@j@jtSTv5H/,S-3:R?r9R}.");
config.setAlgorithm("PBEWITHHMACSHA512ANDAES_256");
config.setKeyObtentionIterations("1000");
config.setPoolSize("1");
config.setProviderName("SunJCE");
config.setSaltGeneratorClassName("org.jasypt.salt.RandomSaltGenerator");
config.setIvGeneratorClassName("org.jasypt.iv.RandomIvGenerator");
config.setStringOutputType("base64");
encryptor.setConfig(config);
System.out.println(encryptor.decrypt("GXKnbq1LS11U2HaONspvH+D/TkIx13aWTaokdkzaF7HSvq6Z0Rv1+JUWFnYopVXu"));
}
}

75
service-executor/pom.xml Normal file
View File

@@ -0,0 +1,75 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.lanyuanxiaoyao</groupId>
<artifactId>hudi-service</artifactId>
<version>1.0.0-SNAPSHOT</version>
</parent>
<artifactId>service-executor</artifactId>
<packaging>pom</packaging>
<modules>
<module>service-executor-core</module>
<module>service-executor-manager</module>
<module>service-executor-task</module>
</modules>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.major.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.major.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-common</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-runtime-blink_${scala.major.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner-blink_${scala.major.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-client</artifactId>
<version>${pulsar.version}</version>
</dependency>
<dependency>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar-client-admin</artifactId>
<version>${pulsar.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-flink${flink.major.version}-bundle</artifactId>
<version>${hudi.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-metrics-prometheus_${scala.major.version}</artifactId>
<version>${flink.version}</version>
</dependency>
</dependencies>
</dependencyManagement>
</project>

14
service-executor/service-executor-core/pom.xml Normal file
View File

@@ -0,0 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.lanyuanxiaoyao</groupId>
<artifactId>service-executor</artifactId>
<version>1.0.0-SNAPSHOT</version>
</parent>
<artifactId>service-executor-core</artifactId>
</project>

12
service-executor/service-executor-core/src/main/java/com/lanyuanxiaoyao/service/executor/core/Task.java Normal file
View File

@@ -0,0 +1,12 @@
package com.lanyuanxiaoyao.service.executor.core;
import java.io.Serializable;
/**
* flink 执行任务信息
*
* @author lanyuanxiaoyao
* @date 2023-12-04
*/
public class Task implements Serializable {
}

12
service-executor/service-executor-core/src/main/java/com/lanyuanxiaoyao/service/executor/core/TaskConstants.java Normal file
View File

@@ -0,0 +1,12 @@
package com.lanyuanxiaoyao.service.executor.core;
/**
* 一些字符串
*
* @author lanyuanxiaoyao
* @date 2023-12-04
*/
public interface TaskConstants {
String TASK_CONTEXT = "task-context";
String TASK_CONTEXT_OPTION = "-" + TASK_CONTEXT;
}

32
service-executor/service-executor-core/src/main/java/com/lanyuanxiaoyao/service/executor/core/TaskContext.java Normal file
View File

@@ -0,0 +1,32 @@
package com.lanyuanxiaoyao.service.executor.core;
import java.io.Serializable;
import java.util.Map;
/**
* flink 任务信息上下文
*
* @author lanyuanxiaoyao
* @date 2023-12-04
*/
public class TaskContext implements Serializable {
private Map<String, Object> metadata;
public TaskContext() {
}
public TaskContext(Map<String, Object> metadata) {
this.metadata = metadata;
}
public Map<String, Object> getMetadata() {
return metadata;
}
@Override
public String toString() {
return "TaskContext{" +
"metadata=" + metadata +
'}';
}
}

161
service-executor/service-executor-manager/pom.xml Normal file
View File

@@ -0,0 +1,161 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.lanyuanxiaoyao</groupId>
<artifactId>service-executor</artifactId>
<version>1.0.0-SNAPSHOT</version>
</parent>
<artifactId>service-executor-manager</artifactId>
<dependencies>
<dependency>
<groupId>com.lanyuanxiaoyao</groupId>
<artifactId>service-executor-core</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.lanyuanxiaoyao</groupId>
<artifactId>service-dependencies</artifactId>
<version>1.0.0-SNAPSHOT</version>
<exclusions>
<exclusion>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.lanyuanxiaoyao</groupId>
<artifactId>service-configuration</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
<exclusions>
<exclusion>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-to-slf4j</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.sun.jersey</groupId>
<artifactId>jersey-client</artifactId>
<version>1.19.4</version>
</dependency>
<dependency>
<groupId>com.sun.jersey</groupId>
<artifactId>jersey-core</artifactId>
<version>1.19.4</version>
</dependency>
<dependency>
<groupId>com.sun.jersey.contribs</groupId>
<artifactId>jersey-apache-client4</artifactId>
<version>1.19.4</version>
</dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-flink${flink.major.version}-bundle</artifactId>
<version>${hudi.version}</version>
</dependency>
<dependency>
<groupId>com.eshore.odcp.hudi.connector</groupId>
<artifactId>executor</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-resources-plugin</artifactId>
<version>3.2.0</version>
<executions>
<execution>
<id>copy-config-file</id>
<phase>validate</phase>
<goals>
<goal>copy-resources</goal>
</goals>
<configuration>
<outputDirectory>${project.build.directory}/classes</outputDirectory>
<resources>
<resource>
<directory>${project.parent.parent.basedir}/config/${build-tag}</directory>
<includes>
<include>*.xml</include>
</includes>
</resource>
</resources>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.3.0</version>
<configuration>
<createDependencyReducedPom>false</createDependencyReducedPom>
<promoteTransitiveDependencies>true</promoteTransitiveDependencies>
<transformers>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/spring.handlers</resource>
</transformer>
<transformer
implementation="org.springframework.boot.maven.PropertiesMergingResourceTransformer">
<resource>META-INF/spring.factories</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/spring.schemas</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.lanyuanxiaoyao.service.executor.manager.ExecutorManagerApplication</mainClass>
</transformer>
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>reference.conf</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
</transformers>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
<exclude>log4j-surefire*.properties</exclude>
</excludes>
</filter>
</filters>
</configuration>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
</execution>
</executions>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>${spring-boot.version}</version>
</dependency>
</dependencies>
</plugin>
</plugins>
</build>
</project>

97
service-executor/service-executor-manager/src/main/java/com/lanyuanxiaoyao/service/executor/manager/ExecutorManagerApplication.java Normal file
View File

@@ -0,0 +1,97 @@
package com.lanyuanxiaoyao.service.executor.manager;
import com.eshore.odcp.hudi.connector.utils.executor.Runner;
import com.lanyuanxiaoyao.service.executor.core.TaskConstants;
import com.ulisesbocchio.jasyptspringboot.annotation.EnableEncryptableProperties;
import java.time.Duration;
import java.util.ArrayList;
import org.apache.flink.client.cli.ClientOptions;
import org.apache.flink.configuration.*;
import org.apache.flink.yarn.configuration.YarnConfigOptions;
import org.apache.flink.yarn.configuration.YarnDeploymentTarget;
import org.apache.hadoop.yarn.api.records.ApplicationId;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.gson.GsonAutoConfiguration;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.retry.annotation.EnableRetry;
import static com.eshore.odcp.hudi.connector.Constants.HALF_HOUR;
import static com.eshore.odcp.hudi.connector.Constants.MINUTE;
/**
* @author lanyuanxiaoyao
* @date 2023-12-04
*/
@EnableDiscoveryClient
@SpringBootApplication(
scanBasePackages = {"com.lanyuanxiaoyao.service"},
exclude = {
GsonAutoConfiguration.class,
DataSourceAutoConfiguration.class
}
)
@EnableConfigurationProperties
@EnableEncryptableProperties
@EnableRetry
public class ExecutorManagerApplication implements ApplicationRunner {
public static void main(String[] args) {
SpringApplication.run(ExecutorManagerApplication.class, args);
}
@Override
public void run(ApplicationArguments args) throws Exception {
Configuration configuration = new Configuration();
configuration.setBoolean(JobManagerOptions.JVM_DIRECT_MEMORY_LIMIT_ENABLED, true);
configuration.setString(AkkaOptions.ASK_TIMEOUT, "10 min");
configuration.setString(AkkaOptions.TCP_TIMEOUT, "15 min");
configuration.setString(AkkaOptions.LOOKUP_TIMEOUT, "10 min");
configuration.set(ClientOptions.CLIENT_TIMEOUT, Duration.ofMinutes(30));
// Kerberos认证
configuration.setBoolean(SecurityOptions.KERBEROS_LOGIN_USETICKETCACHE, true);
configuration.setString(SecurityOptions.KERBEROS_LOGIN_KEYTAB, "/etc/security/keytabs/datalake.app.keytab");
configuration.setString(SecurityOptions.KERBEROS_LOGIN_PRINCIPAL, "datalake/b5s119.hdp.dc@ECLD.COM");
configuration.setLong(HeartbeatManagerOptions.HEARTBEAT_INTERVAL, MINUTE);
configuration.setLong(HeartbeatManagerOptions.HEARTBEAT_TIMEOUT, HALF_HOUR);
configuration.setString(AkkaOptions.ASK_TIMEOUT, "1 min");
configuration.setString(AkkaOptions.TCP_TIMEOUT, "2 min");
configuration.setBoolean(CoreOptions.CHECK_LEAKED_CLASSLOADER, false);
configuration.setString(YarnConfigOptions.APPLICATION_ATTEMPTS, "4");
configuration.setString(YarnConfigOptions.STAGING_DIRECTORY, "hdfs://b2/apps/datalake/yarn");
configuration.setString(ResourceManagerOptions.CONTAINERIZED_MASTER_ENV_PREFIX + "MALLOC_ARENA_MAX", "1");
configuration.setString(ResourceManagerOptions.CONTAINERIZED_TASK_MANAGER_ENV_PREFIX + "MALLOC_ARENA_MAX", "1");
configuration.setInteger(RestOptions.PORT, 8081);
configuration.setString(RestOptions.BIND_PORT, "8084-9400");
configuration.setString(DeploymentOptions.TARGET, YarnDeploymentTarget.APPLICATION.getName());
configuration.set(JobManagerOptions.TOTAL_PROCESS_MEMORY, MemorySize.parse("5120m"));
configuration.set(JobManagerOptions.JVM_METASPACE, MemorySize.parse("128m"));
configuration.set(TaskManagerOptions.TOTAL_FLINK_MEMORY, MemorySize.parse("1024m"));
configuration.set(TaskManagerOptions.MANAGED_MEMORY_SIZE, MemorySize.parse("0m"));
configuration.set(TaskManagerOptions.JVM_METASPACE, MemorySize.parse("128m"));
configuration.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 10);
configuration.setString(YarnConfigOptions.APPLICATION_NAME, "HudiService_faee2e95-660d-4b1c-9cec-13473b3cd5b7");
configuration.setString(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_DIRS, "hdfs://b2/apps/flink/completed-jobs/");
configuration.setLong(HistoryServerOptions.HISTORY_SERVER_ARCHIVE_REFRESH_INTERVAL, 10000);
configuration.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, true);
configuration.setString(YarnConfiguration.LOG_AGGREGATION_REMOTE_APP_LOG_DIR_FMT, "/app-logs");
configuration.set(PipelineOptions.JARS, new ArrayList<String>() {{
add("hdfs://b2/apps/datalake/jars/service/service-executor-task.jar");
}});
ApplicationId applicationId = Runner.run(
configuration,
"com.lanyuanxiaoyao.service.executor.task.Hello",
new String[]{
TaskConstants.TASK_CONTEXT_OPTION,
"{\"metadata\": {\"id\":\"faee2e95-660d-4b1c-9cec-13473b3cd5b7\"}}"
}
);
System.out.println(applicationId);
}
}

5
service-executor/service-executor-manager/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,5 @@
spring:
application:
name: service-executor-manager
profiles:
include: random-port,common,discovery,metrics

51
service-executor/service-executor-manager/src/main/resources/logback-spring.xml Normal file
View File

@@ -0,0 +1,51 @@
<configuration>
<conversionRule conversionWord="clr" converterClass="org.springframework.boot.logging.logback.ColorConverter" />
<conversionRule conversionWord="wex" converterClass="org.springframework.boot.logging.logback.WhitespaceThrowableProxyConverter" />
<conversionRule conversionWord="wEx" converterClass="org.springframework.boot.logging.logback.ExtendedWhitespaceThrowableProxyConverter" />
<springProperty scope="context" name="LOKI_PUSH_URL" source="loki.url"/>
<springProperty scope="context" name="LOGGING_PARENT" source="logging.parent"/>
<springProperty scope="context" name="APP_NAME" source="spring.application.name"/>
<appender name="Loki" class="com.github.loki4j.logback.Loki4jAppender">
<metricsEnabled>true</metricsEnabled>
<http class="com.github.loki4j.logback.ApacheHttpSender">
<url>${LOKI_PUSH_URL:-http://localhost/loki/api/v1/push}</url>
</http>
<format>
<label>
<pattern>app=${APP_NAME:- },host=${HOSTNAME},level=%level</pattern>
</label>
<message>
<pattern>${FILE_LOG_PATTERN:-%d{${LOG_DATEFORMAT_PATTERN:-yyyy-MM-dd HH:mm:ss.SSS}} [${HOSTNAME}] ${LOG_LEVEL_PATTERN:-%5p} ${PID:- } -&#45;&#45; [%t] %-40.40logger{39} #@# : %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}</pattern>
</message>
<sortByTime>true</sortByTime>
</format>
</appender>
<appender name="Console" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>${CONSOLE_LOG_PATTERN:-%clr(%d{${LOG_DATEFORMAT_PATTERN:-yyyy-MM-dd HH:mm:ss.SSS}}){faint} %clr(${LOG_LEVEL_PATTERN:-%5p}) %clr(${PID:- }){magenta} %clr(---){faint} %clr([%15.15t]){faint} %clr(%-40.40logger{39}){cyan} %clr(:){faint} %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}</pattern>
</encoder>
</appender>
<appender name="RollingFile" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${LOGGING_PARENT:-.}/${APP_NAME:-run}.log</file>
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<fileNamePattern>${LOGGING_PARENT:-.}/archive/${APP_NAME:-run}-%d{yyyy-MM-dd}.gz</fileNamePattern>
<MaxHistory>7</MaxHistory>
</rollingPolicy>
<encoder>
<pattern>${FILE_LOG_PATTERN:-%d{${LOG_DATEFORMAT_PATTERN:-yyyy-MM-dd HH:mm:ss.SSS}} [${HOSTNAME}] ${LOG_LEVEL_PATTERN:-%5p} ${PID:- } -&#45;&#45; [%t] %-40.40logger{39} #@# : %m%n${LOG_EXCEPTION_CONVERSION_WORD:-%wEx}}</pattern>
</encoder>
</appender>
<logger name="com.zaxxer.hikari" level="ERROR"/>
<logger name="com.netflix.discovery.shared.resolver.aws.ConfigClusterResolver" level="WARN"/>
<root level="INFO">
<appender-ref ref="Loki"/>
<appender-ref ref="Console"/>
<appender-ref ref="RollingFile"/>
</root>
</configuration>

110
service-executor/service-executor-task/pom.xml Normal file
View File

@@ -0,0 +1,110 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.lanyuanxiaoyao</groupId>
<artifactId>service-executor</artifactId>
<version>1.0.0-SNAPSHOT</version>
</parent>
<artifactId>service-executor-task</artifactId>
<dependencies>
<dependency>
<groupId>com.lanyuanxiaoyao</groupId>
<artifactId>service-executor-core</artifactId>
<version>1.0.0-SNAPSHOT</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.lanyuanxiaoyao</groupId>
<artifactId>service-configuration</artifactId>
<version>1.0.0-SNAPSHOT</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-flink${flink.major.version}-bundle</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_${scala.major.version}</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.major.version}</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-common</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-runtime-blink_${scala.major.version}</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner-blink_${scala.major.version}</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-shaded-guava</artifactId>
<version>30.1.1-jre-15.0</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.3.0</version>
<configuration>
<createDependencyReducedPom>false</createDependencyReducedPom>
<promoteTransitiveDependencies>true</promoteTransitiveDependencies>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>reference.conf</resource>
</transformer>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
</transformers>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
<exclude>log4j-surefire*.properties</exclude>
</excludes>
</filter>
</filters>
</configuration>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

50
service-executor/service-executor-task/src/main/java/com/lanyuanxiaoyao/service/executor/task/Hello.java Normal file
View File

@@ -0,0 +1,50 @@
package com.lanyuanxiaoyao.service.executor.task;
import com.lanyuanxiaoyao.service.executor.core.TaskContext;
import com.lanyuanxiaoyao.service.executor.task.helper.ArgumentsHelper;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Hello world
*
* @author lanyuanxiaoyao
* @date 2023-12-04
*/
public class Hello {
private static final Logger logger = LoggerFactory.getLogger(Hello.class);
public static void main(String[] args) throws Exception {
TaskContext taskContext = ArgumentsHelper.getContext(args);
logger.info("Context: {}", taskContext);
StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
environment
.addSource(new SourceFunction<Integer>() {
@Override
public void run(SourceContext<Integer> context) {
for (int index = 0; index < 10; index++) {
context.collect(index);
}
}
@Override
public void cancel() {
}
})
.map(value -> "Index: " + value)
.addSink(new SinkFunction<String>() {
@Override
public void invoke(String value, Context context) throws Exception {
logger.info("Value: {}", value);
}
});
environment.execute("Service task: Hello");
}
}

23
service-executor/service-executor-task/src/main/java/com/lanyuanxiaoyao/service/executor/task/helper/ArgumentsHelper.java Normal file
View File

@@ -0,0 +1,23 @@
package com.lanyuanxiaoyao.service.executor.task.helper;
import cn.hutool.core.util.StrUtil;
import com.lanyuanxiaoyao.service.executor.core.TaskConstants;
import com.lanyuanxiaoyao.service.executor.core.TaskContext;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonProcessingException;
/**
* 入参解析相关内容
*
* @author ZhangJiacheng
* @date 2022-03-10
*/
public class ArgumentsHelper {
public static TaskContext getContext(String[] args) throws JsonProcessingException {
ParameterTool argsTool = ParameterTool.fromArgs(args);
if (!argsTool.has(TaskConstants.TASK_CONTEXT)) {
throw new RuntimeException(StrUtil.format("Miss argument: {}", TaskConstants.TASK_CONTEXT));
}
return JacksonHelper.getMapper().readValue(argsTool.get(TaskConstants.TASK_CONTEXT), TaskContext.class);
}
}

31
service-executor/service-executor-task/src/main/java/com/lanyuanxiaoyao/service/executor/task/helper/JacksonHelper.java Normal file
View File

@@ -0,0 +1,31 @@
package com.lanyuanxiaoyao.service.executor.task.helper;
import cn.hutool.core.util.ObjectUtil;
import java.io.Serializable;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.DeserializationFeature;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.MapperFeature;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Json 解析相关工具
*
* @author ZhangJiacheng
* @date 2022-06-12
*/
public class JacksonHelper implements Serializable {
private static final Logger logger = LoggerFactory.getLogger(JacksonHelper.class);
private static ObjectMapper INSTANCE = null;
public static ObjectMapper getMapper() {
if (ObjectUtil.isNull(INSTANCE)) {
INSTANCE = new ObjectMapper();
INSTANCE.configure(MapperFeature.ACCEPT_CASE_INSENSITIVE_PROPERTIES, true);
INSTANCE.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
INSTANCE.configure(DeserializationFeature.FAIL_ON_IGNORED_PROPERTIES, false);
}
return INSTANCE;
}
}