flink連接消費kafka實例

這篇文章主要講解了“flink連接消費kafka實例”，文中的講解內容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“flink連接消費kafka實例”吧！

package flink.streaming
import java.util.Properties
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.windowing.time.Time
object StreamingTest {
  def main(args: Array[String]): Unit = {
    val kafkaProps = new Properties()
    //kafka的一些屬性
    kafkaProps.setProperty("bootstrap.servers", "bigdata01:9092")
    //所在的消費組
    kafkaProps.setProperty("group.id", "group1")
    //獲取當前的執行環境
    val evn = StreamExecutionEnvironment.getExecutionEnvironment
    //kafka的consumer，test1是要消費的topic
    val kafkaSource = new FlinkKafkaConsumer[String]("test1",new SimpleStringSchema,kafkaProps)
    //設置從最新的offset開始消費
    kafkaSource.setStartFromLatest()
    //自動提交offset
    kafkaSource.setCommitOffsetsOnCheckpoints(true)
    
    //flink的checkpoint的時間間隔
    evn.enableCheckpointing(5000)
    //添加consumer
    val stream = evn.addSource(kafkaSource)
    stream.setParallelism(3)
    val text = stream.flatMap{ _.toLowerCase().split("\\W+")filter { _.nonEmpty} }
          .map{(_,1)}
          .keyBy(0)
          .timeWindow(Time.seconds(5))
          .sum(1)
         text.print()
     //啟動執行    
    evn.execute("kafkawd")                    
  }
}

//

pom.xml
<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>
  <groupId>hgs</groupId>
  <artifactId>flink_lesson</artifactId>
  <version>1.0.0</version>
  <packaging>jar</packaging>
  <name>flink_lesson</name>
  <url>http://maven.apache.org</url>
  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
  </properties>
  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>4.1</version>
      <scope>test</scope>
    </dependency>
    
    <!-- https://mvnrepository.com/artifact/org.apache.flink/flink-core -->
	<dependency>
	    <groupId>org.apache.flink</groupId>
	    <artifactId>flink-core</artifactId>
	    <version>1.7.1</version>
	</dependency>
	
	<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients -->
	<dependency>
	    <groupId>org.apache.flink</groupId>
	    <artifactId>flink-clients_2.12</artifactId>
	    <version>1.7.1</version>
	</dependency>
	
	<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-streaming-scala -->
	<dependency>
	    <groupId>org.apache.flink</groupId>
	    <artifactId>flink-streaming-scala_2.12</artifactId>
	    <version>1.7.1</version>
	</dependency>
	<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-connector-kafka-0.11 -->
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka_2.12</artifactId>
  <version>1.7.1</version>
</dependency>
	
	<!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
	<dependency>
	    <groupId>io.netty</groupId>
	    <artifactId>netty-all</artifactId>
	    <version>4.1.32.Final</version>
	</dependency>	
  </dependencies>
  
  
   <build>
        <plugins>
            <plugin>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>2.6</version>
                <configuration>
           
                  <archive>
                        <manifest>
                            <!-- 我運行這個jar所運行的主類 -->
                            <mainClass>hgs.flink_lesson.WordCount</mainClass>
                        </manifest>
                    </archive> 
                    
                    <descriptorRefs>
                        <descriptorRef>
                            <!-- 必須是這樣寫 -->
                            jar-with-dependencies
                        </descriptorRef>
                    </descriptorRefs>
                </configuration>
                
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            
              <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin> 
             <plugin>
				<groupId>net.alchim31.maven</groupId>
				<artifactId>scala-maven-plugin</artifactId>
				<version>3.2.0</version>
				<executions>
					<execution>
						<goals>
							<goal>compile</goal>
							<goal>testCompile</goal>
					    </goals>
						<configuration>
							<args>
							<!-- <arg>-make:transitive</arg> -->
                			<arg>-dependencyfile</arg>
                			<arg>${project.build.directory}/.scala_dependencies</arg>
              				</args>
						</configuration>
					</execution>
				</executions>
			</plugin>
			
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-surefire-plugin</artifactId>
				<version>2.18.1</version>
				<configuration>
				<useFile>false</useFile>
				<disableXmlReport>true</disableXmlReport>
				<!-- If you have classpath issue like NoDefClassError,... -->
				<!-- useManifestOnlyJar>false</useManifestOnlyJar -->
				<includes>
					<include>**/*Test.*</include>
					<include>**/*Suite.*</include>
				</includes>
				</configuration>
			</plugin>         
        </plugins>
    </build>
</project>

感謝各位的閱讀，以上就是“flink連接消費kafka實例”的內容了，經過本文的學習后，相信大家對flink連接消費kafka實例這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！