flink通过withParameters()方法传递和使用参数

逃离我推掉我的手 2022-10-15 13:59 183阅读 0赞

**目录**

主类

pom.xml

结果输出

--------------------

##  ##

## 主类 ##

package myflink.Parameter;
    
    import org.apache.flink.api.common.functions.RichFilterFunction;
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.api.java.operators.DataSource;
    import org.apache.flink.configuration.Configuration;
    
    /**
     * @author pangsl
     * @date 2021/5/25 16:58
     * @Description flink通过withParameters()方法传递和使用参数
     */
    public class ConfigurationDemo {
        public static void main(String[] args) throws Exception{
            ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
    
            DataSource<Integer> input = env.fromElements(1, 2, 3, 5, 10, 12, 15, 16);
    
            Configuration configuration = new Configuration();
    
            configuration.setInteger("limit",8);
    
            input.filter(new RichFilterFunction<Integer>() {
                private int limit;
    
                @Override
                public void open(Configuration configuration) throws Exception {
                    limit = configuration.getInteger("limit",0);
                }
    
                @Override
                public boolean filter(Integer value) throws Exception {
                    return value > limit;
                }
            }).withParameters(configuration)
                    .print();
    
        }
    }

## pom.xml ##

<?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">
        <parent>
            <artifactId>flinkdemo</artifactId>
            <groupId>com.flink.pangsl</groupId>
            <version>1.0-SNAPSHOT</version>
        </parent>
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>Flink_shizhanpai</artifactId>
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>8</source>
                        <target>8</target>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
    
        <properties>
            <flink.version>1.11.0</flink.version>
            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        </properties>
    
        <dependencies>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-core</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-planner_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-planner-blink_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-scala_2.11</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
    
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-table-common</artifactId>
                <version>${flink.version}</version>
                <scope>provided</scope>
            </dependency>
    
    
    
        </dependencies>
    
    </project>

## 结果输出 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTExMTAzMDE_size_16_color_FFFFFF_t_70]: /images/20221014/1d2d86dc2b0f40d38605ce5262a80298.png