gfwleak/galaxy-tsg-olap-app-recommend
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

初始版本

Browse Source
This commit is contained in:
LAPTOP-CUUVN8AS\wk
2022-02-22 11:47:29 +08:00
parent 23c4bf3fa2
commit d8ceb36b4b
16 changed files with 1373 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
dependency-reduced-pom.xml Normal file
View File

@@ -0,0 +1,100 @@
<?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/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.galaxy.recommend</groupId>
<artifactId>flink-vpn-recommend</artifactId>
<version>21-01-07</version>
<build>
<plugins>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.8.0</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
<useIncrementalCompilation>false</useIncrementalCompilation>
<compilerArgs>
<arg>-Xpkginfo:always</arg>
</compilerArgs>
</configuration>
</plugin>
<plugin>
<artifactId>maven-shade-plugin</artifactId>
<executions>
<execution>
<id>flink-vpn-recommend</id>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<finalName>flink-vpn-recommend-21-01-07</finalName>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF</exclude>
</excludes>
</filter>
</filters>
<transformers>
<transformer>
<mainClass>com.galaxy.recommend.Recommendation</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<id>nexus</id>
<name>Team Nexus Repository</name>
<url>http://192.168.40.125:8099/content/groups/public</url>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.12</artifactId>
<version>1.13.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.13.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
<version>1.13.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId>
<version>1.13.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-csv</artifactId>
<version>1.13.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>1.13.1</version>
<scope>provided</scope>
</dependency>
</dependencies>
<properties>
<hadoop.version>2.7.1</hadoop.version>
<flink.version>1.13.1</flink.version>
</properties>
</project>

171
flink-vpn-recommend.iml Normal file
View File

@@ -0,0 +1,171 @@
<?xml version="1.0" encoding="UTF-8"?>
<module org.jetbrains.idea.maven.project.MavenProjectsManager.isMavenModule="true" type="JAVA_MODULE" version="4">
<component name="NewModuleRootManager" LANGUAGE_LEVEL="JDK_1_8">
<output url="file://$MODULE_DIR$/target/classes" />
<output-test url="file://$MODULE_DIR$/target/test-classes" />
<content url="file://$MODULE_DIR$">
<sourceFolder url="file://$MODULE_DIR$/src/main/java" isTestSource="false" />
<sourceFolder url="file://$MODULE_DIR$/src/main/resources" type="java-resource" />
<sourceFolder url="file://$MODULE_DIR$/src/test/java" isTestSource="true" />
<excludeFolder url="file://$MODULE_DIR$/target" />
</content>
<orderEntry type="inheritedJdk" />
<orderEntry type="sourceFolder" forTests="false" />
<orderEntry type="library" name="Maven: com.zdjizhi:galaxy:1.0.6" level="project" />
<orderEntry type="library" name="Maven: org.apache.commons:commons-lang3:3.4" level="project" />
<orderEntry type="library" name="Maven: commons-io:commons-io:2.4" level="project" />
<orderEntry type="library" name="Maven: log4j:log4j:1.2.14" level="project" />
<orderEntry type="library" name="Maven: joda-time:joda-time:2.10" level="project" />
<orderEntry type="library" name="Maven: com.maxmind.geoip:geoip-api:1.3.1" level="project" />
<orderEntry type="library" name="Maven: com.maxmind.geoip2:geoip2:2.12.0" level="project" />
<orderEntry type="library" name="Maven: org.apache.httpcomponents:httpclient:4.5.5" level="project" />
<orderEntry type="library" name="Maven: org.apache.httpcomponents:httpcore:4.4.9" level="project" />
<orderEntry type="library" name="Maven: com.fasterxml.jackson.core:jackson-databind:2.9.5" level="project" />
<orderEntry type="library" name="Maven: com.fasterxml.jackson.core:jackson-core:2.9.5" level="project" />
<orderEntry type="library" name="Maven: com.fasterxml.jackson.core:jackson-annotations:2.9.5" level="project" />
<orderEntry type="library" name="Maven: com.maxmind.db:maxmind-db:1.2.2" level="project" />
<orderEntry type="library" name="Maven: com.google.guava:guava:23.0" level="project" />
<orderEntry type="library" name="Maven: com.google.errorprone:error_prone_annotations:2.0.18" level="project" />
<orderEntry type="library" name="Maven: com.google.j2objc:j2objc-annotations:1.1" level="project" />
<orderEntry type="library" name="Maven: org.codehaus.mojo:animal-sniffer-annotations:1.14" level="project" />
<orderEntry type="library" name="Maven: cn.hutool:hutool-all:5.5.2" level="project" />
<orderEntry type="library" name="Maven: org.slf4j:slf4j-api:1.7.21" level="project" />
<orderEntry type="library" name="Maven: org.slf4j:slf4j-log4j12:1.7.21" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-streaming-java_2.12:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-file-sink-common:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-runtime_2.12:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-queryable-state-client-java:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-hadoop-fs:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-shaded-netty:4.1.49.Final-13.0" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-shaded-jackson:2.12.1-13.0" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-shaded-zookeeper-3:3.4.14-13.0" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.javassist:javassist:3.24.0-GA" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.scala-lang:scala-library:2.12.7" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe.akka:akka-actor_2.12:2.5.21" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe:config:1.3.3" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.scala-lang.modules:scala-java8-compat_2.12:0.8.0" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe.akka:akka-stream_2.12:2.5.21" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.reactivestreams:reactive-streams:1.0.2" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe:ssl-config-core_2.12:0.3.7" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.scala-lang.modules:scala-parser-combinators_2.12:1.1.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe.akka:akka-protobuf_2.12:2.5.21" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe.akka:akka-slf4j_2.12:2.5.21" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.clapper:grizzled-slf4j_2.12:1.3.2" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.github.scopt:scopt_2.12:3.5.0" level="project" />
<orderEntry type="library" name="Maven: org.xerial.snappy:snappy-java:1.1.8.3" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.twitter:chill_2.12:0.7.6" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.twitter:chill-java:0.7.6" level="project" />
<orderEntry type="library" name="Maven: org.lz4:lz4-java:1.6.0" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-shaded-guava:18.0-13.0" level="project" />
<orderEntry type="library" name="Maven: org.apache.commons:commons-math3:3.5" level="project" />
<orderEntry type="library" name="Maven: com.google.code.findbugs:jsr305:1.3.9" level="project" />
<orderEntry type="library" name="Maven: org.apache.flink:force-shading:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-java:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-core:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-annotations:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-metrics-core:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-shaded-asm-7:7.1-13.0" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.esotericsoftware.kryo:kryo:2.24.0" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.esotericsoftware.minlog:minlog:1.2" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.objenesis:objenesis:2.1" level="project" />
<orderEntry type="library" name="Maven: commons-collections:commons-collections:3.2.2" level="project" />
<orderEntry type="library" name="Maven: org.apache.commons:commons-compress:1.20" level="project" />
<orderEntry type="library" name="Maven: org.apache.flink:flink-connector-kafka_2.12:1.13.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.kafka:kafka-clients:2.4.1" level="project" />
<orderEntry type="library" name="Maven: com.github.luben:zstd-jni:1.4.3-1" level="project" />
<orderEntry type="library" name="Maven: org.apache.flink:flink-connector-base:1.13.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.zookeeper:zookeeper:3.4.9" level="project" />
<orderEntry type="library" name="Maven: jline:jline:0.9.94" level="project" />
<orderEntry type="library" name="Maven: io.netty:netty:3.10.5.Final" level="project" />
<orderEntry type="library" name="Maven: org.apache.hbase:hbase-client:2.2.3" level="project" />
<orderEntry type="library" name="Maven: org.apache.hbase.thirdparty:hbase-shaded-protobuf:2.2.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hbase:hbase-common:2.2.3" level="project" />
<orderEntry type="library" name="Maven: com.github.stephenc.findbugs:findbugs-annotations:1.3.9-1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hbase:hbase-hadoop-compat:2.2.3" level="project" />
<orderEntry type="library" name="Maven: org.apache.hbase:hbase-metrics-api:2.2.3" level="project" />
<orderEntry type="library" name="Maven: org.apache.hbase:hbase-hadoop2-compat:2.2.3" level="project" />
<orderEntry type="library" name="Maven: org.apache.hbase:hbase-metrics:2.2.3" level="project" />
<orderEntry type="library" name="Maven: org.apache.hbase:hbase-protocol-shaded:2.2.3" level="project" />
<orderEntry type="library" name="Maven: org.apache.hbase:hbase-protocol:2.2.3" level="project" />
<orderEntry type="library" name="Maven: commons-codec:commons-codec:1.10" level="project" />
<orderEntry type="library" name="Maven: org.apache.hbase.thirdparty:hbase-shaded-miscellaneous:2.2.1" level="project" />
<orderEntry type="library" name="Maven: com.google.protobuf:protobuf-java:2.5.0" level="project" />
<orderEntry type="library" name="Maven: org.apache.hbase.thirdparty:hbase-shaded-netty:2.2.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.htrace:htrace-core4:4.2.0-incubating" level="project" />
<orderEntry type="library" name="Maven: org.jruby.jcodings:jcodings:1.0.18" level="project" />
<orderEntry type="library" name="Maven: org.jruby.joni:joni:2.1.11" level="project" />
<orderEntry type="library" name="Maven: io.dropwizard.metrics:metrics-core:3.2.6" level="project" />
<orderEntry type="library" name="Maven: org.apache.commons:commons-crypto:1.0.0" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-auth:2.8.5" level="project" />
<orderEntry type="library" name="Maven: com.nimbusds:nimbus-jose-jwt:4.41.1" level="project" />
<orderEntry type="library" name="Maven: com.github.stephenc.jcip:jcip-annotations:1.0-1" level="project" />
<orderEntry type="library" name="Maven: org.apache.directory.server:apacheds-kerberos-codec:2.0.0-M15" level="project" />
<orderEntry type="library" name="Maven: org.apache.directory.server:apacheds-i18n:2.0.0-M15" level="project" />
<orderEntry type="library" name="Maven: org.apache.directory.api:api-asn1-api:1.0.0-M20" level="project" />
<orderEntry type="library" name="Maven: org.apache.directory.api:api-util:1.0.0-M20" level="project" />
<orderEntry type="library" name="Maven: org.apache.curator:curator-framework:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-common:2.8.5" level="project" />
<orderEntry type="library" name="Maven: xmlenc:xmlenc:0.52" level="project" />
<orderEntry type="library" name="Maven: commons-net:commons-net:3.1" level="project" />
<orderEntry type="library" name="Maven: org.mortbay.jetty:jetty-sslengine:6.1.26" level="project" />
<orderEntry type="library" name="Maven: commons-logging:commons-logging:1.1.3" level="project" />
<orderEntry type="library" name="Maven: commons-lang:commons-lang:2.6" level="project" />
<orderEntry type="library" name="Maven: commons-configuration:commons-configuration:1.6" level="project" />
<orderEntry type="library" name="Maven: commons-digester:commons-digester:1.8" level="project" />
<orderEntry type="library" name="Maven: commons-beanutils:commons-beanutils:1.7.0" level="project" />
<orderEntry type="library" name="Maven: commons-beanutils:commons-beanutils-core:1.8.0" level="project" />
<orderEntry type="library" name="Maven: org.codehaus.jackson:jackson-core-asl:1.9.13" level="project" />
<orderEntry type="library" name="Maven: org.codehaus.jackson:jackson-mapper-asl:1.9.13" level="project" />
<orderEntry type="library" name="Maven: org.apache.avro:avro:1.7.4" level="project" />
<orderEntry type="library" name="Maven: com.thoughtworks.paranamer:paranamer:2.3" level="project" />
<orderEntry type="library" name="Maven: com.google.code.gson:gson:2.2.4" level="project" />
<orderEntry type="library" name="Maven: com.jcraft:jsch:0.1.54" level="project" />
<orderEntry type="library" name="Maven: org.apache.curator:curator-client:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.curator:curator-recipes:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.yetus:audience-annotations:0.5.0" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-client:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-hdfs:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.mortbay.jetty:jetty-util:6.1.26" level="project" />
<orderEntry type="library" name="Maven: io.netty:netty-all:4.0.23.Final" level="project" />
<orderEntry type="library" name="Maven: xerces:xercesImpl:2.9.1" level="project" />
<orderEntry type="library" name="Maven: xml-apis:xml-apis:1.3.04" level="project" />
<orderEntry type="library" name="Maven: org.apache.htrace:htrace-core:3.1.0-incubating" level="project" />
<orderEntry type="library" name="Maven: org.fusesource.leveldbjni:leveldbjni-all:1.8" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-mapreduce-client-app:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-mapreduce-client-common:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-yarn-client:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-yarn-server-common:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-mapreduce-client-shuffle:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-yarn-api:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-mapreduce-client-core:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-yarn-common:2.7.1" level="project" />
<orderEntry type="library" name="Maven: javax.xml.bind:jaxb-api:2.2.2" level="project" />
<orderEntry type="library" name="Maven: javax.xml.stream:stax-api:1.0-2" level="project" />
<orderEntry type="library" name="Maven: javax.activation:activation:1.1" level="project" />
<orderEntry type="library" name="Maven: com.sun.jersey:jersey-core:1.9" level="project" />
<orderEntry type="library" name="Maven: com.sun.jersey:jersey-client:1.9" level="project" />
<orderEntry type="library" name="Maven: org.codehaus.jackson:jackson-jaxrs:1.9.13" level="project" />
<orderEntry type="library" name="Maven: org.codehaus.jackson:jackson-xc:1.9.13" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-mapreduce-client-jobclient:2.7.1" level="project" />
<orderEntry type="library" name="Maven: org.apache.hadoop:hadoop-annotations:2.7.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-json:1.13.1" level="project" />
<orderEntry type="library" name="Maven: com.alibaba:fastjson:1.2.70" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-csv:1.13.1" level="project" />
<orderEntry type="library" name="Maven: com.opencsv:opencsv:3.3" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-clients_2.11:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-runtime_2.11:1.13.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe.akka:akka-actor_2.11:2.5.21" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.scala-lang.modules:scala-java8-compat_2.11:0.7.0" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe.akka:akka-stream_2.11:2.5.21" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe:ssl-config-core_2.11:0.3.7" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.scala-lang.modules:scala-parser-combinators_2.11:1.1.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe.akka:akka-protobuf_2.11:2.5.21" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.typesafe.akka:akka-slf4j_2.11:2.5.21" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.clapper:grizzled-slf4j_2.11:1.3.2" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.github.scopt:scopt_2.11:3.5.0" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: com.twitter:chill_2.11:0.7.6" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-optimizer_2.11:1.13.1" level="project" />
<orderEntry type="library" name="Maven: commons-cli:commons-cli:1.3.1" level="project" />
<orderEntry type="library" scope="PROVIDED" name="Maven: org.apache.flink:flink-streaming-java_2.11:1.13.1" level="project" />
</component>
</module>

240
pom.xml Normal file
View File

@@ -0,0 +1,240 @@
<?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>
<groupId>com.galaxy.recommend</groupId>
<artifactId>flink-vpn-recommend</artifactId>
<version>21-01-07</version>
<repositories>
<repository>
<id>nexus</id>
<name>Team Nexus Repository</name>
<url>http://192.168.40.125:8099/content/groups/public</url>
</repository>
</repositories>
<properties>
<flink.version>1.13.1</flink.version>
<hadoop.version>2.7.1</hadoop.version>
</properties>
<dependencies>
<dependency>
<groupId>com.zdjizhi</groupId>
<artifactId>galaxy</artifactId>
<version>1.0.6</version>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>log4j-over-slf4j</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>5.5.2</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.21</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.21</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.12</artifactId>
<version>1.13.1</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<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-connector-kafka_2.12</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.9</version>
</dependency>
<dependency>
<groupId>org.apache.hbase</groupId>
<artifactId>hbase-client</artifactId>
<version>2.2.3</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>2.7.1</version>
<exclusions>
<exclusion>
<artifactId>slf4j-log4j12</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>log4j-over-slf4j</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
<exclusion>
<artifactId>servlet-api</artifactId>
<groupId>javax.servlet</groupId>
</exclusion>
</exclusions>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table -->
<!-- <dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table</artifactId>
<version>${flink.version}</version>
<type>pom</type>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-sql-connector-kafka_2.11</artifactId>
<version>${flink.version}</version>
</dependency>-->
<!-- https://mvnrepository.com/artifact/org.apache.flink/flink-json -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.70</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-csv</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.opencsv</groupId>
<artifactId>opencsv</artifactId>
<version>3.3</version>
</dependency>
<!-- Flink modules -->
<!-- <dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java</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-table-planner_2.11</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>-->
<!-- CLI dependencies -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>${flink.version}</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.8.0</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
<!-- The semantics of this option are reversed, see MCOMPILER-209. -->
<useIncrementalCompilation>false</useIncrementalCompilation>
<compilerArgs>
<!-- Prevents recompilation due to missing package-info.class, see MCOMPILER-205 -->
<arg>-Xpkginfo:always</arg>
</compilerArgs>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<executions>
<execution>
<id>flink-vpn-recommend</id>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<finalName>flink-vpn-recommend-21-01-07</finalName>
<filters>
<filter>
<!-- Do not copy the signatures in the META-INF folder.
Otherwise, this might cause SecurityExceptions when using the JAR. -->
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF</exclude>
</excludes>
</filter>
</filters>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.galaxy.recommend.Recommendation</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>

37
src/main/java/com/galaxy/cn/common/CipEntity.java Normal file
View File

@@ -0,0 +1,37 @@
package com.galaxy.cn.common;
public class CipEntity {
public long common_recv_time ;
public String common_client_list ;
public String common_app_label ;
public long getCommon_recv_time() {
return common_recv_time;
}
public void setCommon_recv_time(long common_recv_time) {
this.common_recv_time = common_recv_time;
}
public String getCommon_client_list() {
return common_client_list;
}
public void setCommon_client_list(String common_client_list) {
this.common_client_list = common_client_list;
}
public String getCommon_app_label() {
return common_app_label;
}
public void setCommon_app_label(String common_app_label) {
this.common_app_label = common_app_label;
}
public CipEntity() {
}
}

55
src/main/java/com/galaxy/cn/common/Entity.java Normal file
View File

@@ -0,0 +1,55 @@
package com.galaxy.cn.common;
import java.io.Serializable;
public class Entity implements Serializable {
public int ifError;
public long common_processing_time ;
public String common_client_ip ;
public String common_app_label ;
public long common_sessions;
public Entity() {
}
public int getIfError() {
return ifError;
}
public void setIfError(int ifError) {
this.ifError = ifError;
}
public long getCommon_recv_time() {
return common_processing_time;
}
public String getCommon_app_label() {
return common_app_label;
}
public void setCommon_app_label(String common_app_label) {
this.common_app_label = common_app_label;
}
public void setCommon_recv_time(long common_recv_time) {
this.common_processing_time = common_recv_time;
}
public String getCommon_client_ip() {
return common_client_ip;
}
public void setCommon_client_ip(String common_client_ip) {
this.common_client_ip = common_client_ip;
}
public long getCommon_sessions() {
return common_sessions;
}
public void setCommon_sessions(long common_sessions) {
this.common_sessions = common_sessions;
}
}

55
src/main/java/com/galaxy/cn/common/ObjectEntity.java Normal file
View File

@@ -0,0 +1,55 @@
package com.galaxy.cn.common;
public class ObjectEntity implements Comparable<ObjectEntity> {
public String common_client_ip ;
public String common_app_label ;
public long sessions ;
public long common_recv_time ;
public long getCommon_recv_time() {
return common_recv_time;
}
public void setCommon_recv_time(long common_recv_time) {
this.common_recv_time = common_recv_time;
}
public String getCommon_client_ip() {
return common_client_ip;
}
public void setCommon_client_ip(String common_client_ip) {
this.common_client_ip = common_client_ip;
}
public String getCommon_app_label() {
return common_app_label;
}
public void setCommon_app_label(String common_app_label) {
this.common_app_label = common_app_label;
}
public long getSessions() {
return sessions;
}
public void setSessions(long sessions) {
this.sessions = sessions;
}
@Override
public int compareTo(ObjectEntity per) {
if(this.sessions<per.sessions){
return -1 ;
}else if(this.sessions>per.sessions){
return 1 ;
}else{
return this.common_client_ip.compareTo(per.common_client_ip) ; // 调用String中的compareTo()方法
}
}
}

57
src/main/java/com/galaxy/cn/common/ResultEntity.java Normal file
View File

@@ -0,0 +1,57 @@
package com.galaxy.cn.common;
public class ResultEntity implements Comparable<ResultEntity> {
public String common_client_ip ;
public long sessions ;
/*
public String common_app_label ;
public long common_recv_time ;
*/
/* public String getCommon_app_label() {
return common_app_label;
}
public void setCommon_app_label(String common_app_label) {
this.common_app_label = common_app_label;
}
public long getCommon_recv_time() {
return common_recv_time;
}
public void setCommon_recv_time(long common_recv_time) {
this.common_recv_time = common_recv_time;
}*/
public String getCommon_client_ip() {
return common_client_ip;
}
public void setCommon_client_ip(String common_client_ip) {
this.common_client_ip = common_client_ip;
}
public long getSessions() {
return sessions;
}
public void setSessions(long sessions) {
this.sessions = sessions;
}
@Override
public int compareTo(ResultEntity per) {
if(this.sessions<per.sessions){
return -1 ;
}else if(this.sessions>per.sessions){
return 1 ;
}else{
return this.common_client_ip.compareTo(per.common_client_ip) ; // 调用String中的compareTo()方法
}
}
}

62
src/main/java/com/galaxy/cn/config/CommonConfigurations.java Normal file
View File

@@ -0,0 +1,62 @@
package com.galaxy.cn.config;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
/**
* @author Administrator
*/
public final class CommonConfigurations {
private static Properties propService = new Properties();
public static Map<String,String> getHashTableProperty(String key) {
Map<String,String> map = new HashMap<>();
String[] keyarray = propService.getProperty(key).split(",");
for(String k :keyarray){
if(k!=null && !"".equals(k.trim())){
map.put(k,"");
}
}
return map;
}
public static String getStringProperty(String key) {
return propService.getProperty(key);
}
public static Integer getIntProperty( String key) {
return Integer.parseInt(propService.getProperty(key));
}
public static Long getLongProperty(String key) {
return Long.parseLong(propService.getProperty(key));
}
public static Boolean getBooleanProperty(String key) {
return "true".equals(propService.getProperty(key).toLowerCase().trim());
}
static {
try {
propService.load(CommonConfigurations.class.getClassLoader().getResourceAsStream("common.properties"));
} catch (Exception e) {
propService = null;
}
}
}

40
src/main/java/com/galaxy/cn/config/commonConfig.java Normal file
View File

@@ -0,0 +1,40 @@
package com.galaxy.cn.config;
import java.util.Map;
/**
* Created by wk on 2021/1/6.
*/
public class commonConfig {
public static final String SOURCE_KAFKA_BROKER = CommonConfigurations.getStringProperty("source.kafka.broker");
public static final String SOURCE_KAFKA_GROUP_ID = CommonConfigurations.getStringProperty("source.kafka.group.id");
public static final String SOURCE_KAFKA_TOPIC = CommonConfigurations.getStringProperty("source.kafka.topic");
public static final String ZK_HOST = CommonConfigurations.getStringProperty("zk.host");
public static final int SOURCE_KAFKA_PARALLELISM = CommonConfigurations.getIntProperty("source.kafka.parallelism");
public static final int SINK_HBASE_PARALLELISM = CommonConfigurations.getIntProperty("sink.hbase.parallelism");
public static final String SINK_HBASE_FM = CommonConfigurations.getStringProperty("sink.hbase.fm");
public static final String SINK_HBASE_TABLE = CommonConfigurations.getStringProperty("sink.hbase.table");
public static final int TASK_PARALLELISM = CommonConfigurations.getIntProperty("task.parallelism");
public static final int WATERMARK_TIME = CommonConfigurations.getIntProperty("watermark.time");
public static final int SLIDINGWINDOW_TIME_MINUTE = CommonConfigurations.getIntProperty("slidingwindow.time.minute");
public static final int SLIDINGWINDOWSLOT_TIME_MINUTE = CommonConfigurations.getIntProperty("slidingwindowslot.time.minute");
public static final int TOP_LIMIT = CommonConfigurations.getIntProperty("top.limit");
public static final String KAFKA_USER = CommonConfigurations.getStringProperty("kafka.user");
public static final String KAFKA_PIN = CommonConfigurations.getStringProperty("kafka.pin");
public static final int KAFKA_SECURITY = CommonConfigurations.getIntProperty("kafka.security");
public static final String TOOLS_LIBRARY = CommonConfigurations.getStringProperty("tools.library");
public static final Boolean no_filter = CommonConfigurations.getBooleanProperty("has.filter");
public static final Map<String,String> app_white_list= CommonConfigurations.getHashTableProperty("app.white.list");
public static final String SESSION_TIMEOUT_MS=CommonConfigurations.getStringProperty("session.timeout.ms");
public static final String MAX_POLL_RECORD=CommonConfigurations.getStringProperty("max.poll.records");
public static final String MAX_PARTITION_FETCH_BYTES=CommonConfigurations.getStringProperty("max.partition.fetch.bytes");
}

137
src/main/java/com/galaxy/cn/function/TopNHotItems.java Normal file
View File

@@ -0,0 +1,137 @@
package com.galaxy.cn.function;
import com.alibaba.fastjson.JSONObject;
import com.galaxy.cn.common.ObjectEntity;
import com.galaxy.cn.common.ResultEntity;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.util.*;
public class TopNHotItems extends KeyedProcessFunction<Tuple1<String>, ObjectEntity, Tuple3<String,String,Long>> {
private final int topSize;
// Set<ObjectEntity> allSet = new TreeSet<ObjectEntity>() ;
private Map<String,PriorityQueue<ResultEntity>> resultMap ;
private Map<String,Map<String,Long>> countMap ;
public TopNHotItems(int i) {
this.topSize = i;
}
@Override
public void open(Configuration parameters) throws Exception {
this.resultMap= new HashMap<>();
this.countMap= new HashMap<>();
}
@Override
public void processElement(ObjectEntity objectEntity, Context context, Collector<Tuple3<String,String,Long>> collector) throws Exception {
//allSet.add(objectEntity);
if(countMap.containsKey(objectEntity.getCommon_app_label())) {
PriorityQueue<ResultEntity> queue =resultMap.get(objectEntity.getCommon_app_label());
Map<String,Long> map = countMap.get(objectEntity.getCommon_app_label());
if (map.size() < topSize) {
ResultEntity resultEntity =new ResultEntity();
resultEntity.setSessions(objectEntity.getSessions());
resultEntity.setCommon_client_ip(objectEntity.getCommon_client_ip());
// resultEntity.setCommon_app_label(objectEntity.getCommon_app_label());
// resultEntity.setCommon_recv_time(objectEntity.getCommon_recv_time());
queue.add(resultEntity);
map.put(objectEntity.getCommon_client_ip(),objectEntity.getSessions());
} else {
if (queue.peek() != null) {
ResultEntity res=queue.peek();
if (res.getSessions() <= objectEntity.getSessions()) {
queue.poll();
map.remove(res.getCommon_client_ip());
ResultEntity resultEntity =new ResultEntity();
resultEntity.setSessions(objectEntity.getSessions());
resultEntity.setCommon_client_ip(objectEntity.getCommon_client_ip());
// resultEntity.setCommon_app_label(objectEntity.getCommon_app_label());
// resultEntity.setCommon_recv_time(objectEntity.getCommon_recv_time());
queue.add(resultEntity);
map.put(objectEntity.getCommon_client_ip(),objectEntity.getSessions());
}
}
}
}
else{
PriorityQueue<ResultEntity> que = new PriorityQueue<>();
Map<String,Long> map = new HashMap<>();
ResultEntity resultEntity =new ResultEntity();
resultEntity.setSessions(objectEntity.getSessions());
resultEntity.setCommon_client_ip(objectEntity.getCommon_client_ip());
// resultEntity.setCommon_app_label(objectEntity.getCommon_app_label());
// resultEntity.setCommon_recv_time(objectEntity.getCommon_recv_time());
que.add(resultEntity);
map.put(objectEntity.getCommon_client_ip(),objectEntity.getSessions());
resultMap.put(objectEntity.getCommon_app_label(),que);
countMap.put(objectEntity.getCommon_app_label(),map);
}
//注册 windowEnd+1 的 EventTime Timer, 当触发时，说明收集好了所有 windowEnd的商品数据
context.timerService().registerEventTimeTimer(objectEntity.getCommon_recv_time() + 1);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple3<String,String,Long>> out) throws Exception {
if(resultMap.size()>0) {
/*
List<Map.Entry<String, Long>> objectlist = new ArrayList<Map.Entry<String, Long>>(resultMap.entrySet());
Collections.sort(objectlist, new Comparator<Map.Entry<String, Long>>() {
public int compare(Map.Entry<String, Long> o1, Map.Entry<String, Long> o2) {
return Math.toIntExact((o2.getValue() - o1.getValue()));
}
});
*/
for (Map.Entry<String, PriorityQueue<ResultEntity>> entry : resultMap.entrySet()) {
PriorityQueue<ResultEntity> queue= entry.getValue();
String jsonStr = JSONObject.toJSONString(queue);
Tuple3<String,String,Long> emit= new Tuple3<>(entry.getKey(),jsonStr,ctx.timestamp());
out.collect(emit);
}
resultMap.clear();
countMap.clear();
}
/* if(queue.size()>0) {
int i = 0;
StringBuilder sb = new StringBuilder();
CipEntity ce = new CipEntity();
for (ObjectEntity obj : queue) {
if (i == 0) {
ce.setCommon_app_label(obj.getCommon_app_label());
ce.setCommon_recv_time(obj.getCommon_recv_time());
}
sb.append(obj.getCommon_client_ip());
sb.append(",");
i++;
if (i >= topSize) {
break;
}
}
ce.setCommon_client_list(sb.toString());
out.collect(ce);
queue.clear();
}*/
}
}

98
src/main/java/com/galaxy/cn/function/TopnHotItem.java Normal file
View File

@@ -0,0 +1,98 @@
package com.galaxy.cn.function;
import com.alibaba.fastjson.JSONObject;
import com.galaxy.cn.common.ObjectEntity;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.util.*;
public class TopnHotItem extends KeyedProcessFunction<Tuple1<String>, ObjectEntity, Tuple3<String,String,Long>> {
private final int topSize;
// Set<ObjectEntity> allSet = new TreeSet<ObjectEntity>() ;
private PriorityQueue<ObjectEntity> queue ;
private Map<String,Long> resultMap ;
public TopnHotItem(int i) {
this.topSize = i;
}
@Override
public void open(Configuration parameters) throws Exception {
this.queue= new PriorityQueue<>();
this.resultMap= new HashMap<>();
}
@Override
public void processElement(ObjectEntity objectEntity, Context context, Collector<Tuple3<String,String,Long>> collector) throws Exception {
//allSet.add(objectEntity);
if (resultMap.size() < topSize) {
queue.add(objectEntity);
resultMap.put(objectEntity.getCommon_client_ip(),objectEntity.getSessions());
} else {
if(queue.peek()!=null){
if (queue.peek().getSessions()<objectEntity.getSessions()) {
resultMap.put(objectEntity.getCommon_client_ip(),objectEntity.getSessions());
queue.poll();
queue.add(objectEntity);
}
}
}
//注册 windowEnd+1 的 EventTime Timer, 当触发时，说明收集好了所有 windowEnd的商品数据
context.timerService().registerEventTimeTimer(objectEntity.getCommon_recv_time() + 1);
}
@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple3<String,String,Long>> out) throws Exception {
if(resultMap.size()>0) {
List<Map.Entry<String, Long>> objectlist = new ArrayList<Map.Entry<String, Long>>(resultMap.entrySet());
Collections.sort(objectlist, new Comparator<Map.Entry<String, Long>>() {
public int compare(Map.Entry<String, Long> o1, Map.Entry<String, Long> o2) {
return Math.toIntExact((o2.getValue() - o1.getValue()));
}
});
String jsonStr = JSONObject.toJSONString(objectlist);
Tuple3<String,String,Long> emit= new Tuple3<>(ctx.getCurrentKey().f0,jsonStr,ctx.timestamp());
out.collect(emit);
queue.clear();
resultMap.clear();
}
/* if(queue.size()>0) {
int i = 0;
StringBuilder sb = new StringBuilder();
CipEntity ce = new CipEntity();
for (ObjectEntity obj : queue) {
if (i == 0) {
ce.setCommon_app_label(obj.getCommon_app_label());
ce.setCommon_recv_time(obj.getCommon_recv_time());
}
sb.append(obj.getCommon_client_ip());
sb.append(",");
i++;
if (i >= topSize) {
break;
}
}
ce.setCommon_client_list(sb.toString());
out.collect(ce);
queue.clear();
}*/
}
}

30
src/main/java/com/galaxy/cn/function/metricsCalculate.java Normal file
View File

@@ -0,0 +1,30 @@
package com.galaxy.cn.function;
import com.galaxy.cn.common.Entity;
import com.galaxy.cn.common.ObjectEntity;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
public class metricsCalculate extends ProcessWindowFunction<
Entity, // 输入类型
ObjectEntity, // 输出类型
Tuple2<String,String>, // 键类型
TimeWindow> { // 窗口类型
@Override
public void process(Tuple2<String,String> s,
Context context,
Iterable<Entity> elements, Collector<ObjectEntity> out) throws Exception {
ObjectEntity objEntity= new ObjectEntity();
objEntity.setCommon_recv_time(context.window().getEnd()/1000);
objEntity.setCommon_app_label(s.f0);
objEntity.setCommon_client_ip(s.f1);
for (Entity event : elements) {
objEntity.sessions+=event.common_sessions;
}
out.collect(objEntity);
}
}

67
src/main/java/com/galaxy/cn/sink/HbaseSink.java Normal file
View File

@@ -0,0 +1,67 @@
package com.galaxy.cn.sink;
import com.galaxy.cn.config.commonConfig;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;
import java.io.Serializable;
import static cn.hutool.crypto.SecureUtil.md5;
public class HbaseSink extends RichSinkFunction<Tuple3<String,String,Long>> implements Serializable, SinkFunction<Tuple3<String, String,Long>> {
private Logger log;
private String hbase_zookeeper_host;
private Connection connection;
private Admin admin;
public HbaseSink(String hbase_zookeeper_host) {
this.hbase_zookeeper_host = hbase_zookeeper_host;
}
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
log = Logger.getLogger(HbaseSink.class);
org.apache.hadoop.conf.Configuration configuration = HBaseConfiguration.create();
configuration.set("hbase.zookeeper.quorum", hbase_zookeeper_host);
connection = ConnectionFactory.createConnection(configuration);
admin = connection.getAdmin();
}
public void invoke(Tuple3<String,String,Long> data, Context context) throws Exception {
// 按 project:table 归纳
Table table = null;
try {
table = connection.getTable(TableName.valueOf(commonConfig.SINK_HBASE_TABLE));
Put put = new Put(Bytes.toBytes(md5(data.f0)));
put.addColumn(Bytes.toBytes(commonConfig.SINK_HBASE_FM), Bytes.toBytes("app_label"), Bytes.toBytes(data.f0));
put.addColumn(Bytes.toBytes(commonConfig.SINK_HBASE_FM), Bytes.toBytes("client_ip_list"), Bytes.toBytes(data.f1));
put.addColumn(Bytes.toBytes(commonConfig.SINK_HBASE_FM), Bytes.toBytes("last_update_time"), Bytes.toBytes(data.f2));
table.put(put);
} catch (Exception e) {
log.error(e.toString());
} finally {
table.close();
}
}
@Override
public void close() throws Exception {
super.close();
}
}

143
src/main/java/com/galaxy/recommend/Recommendation.java Normal file
View File

@@ -0,0 +1,143 @@
package com.galaxy.recommend;
import com.alibaba.fastjson.JSON;
import com.galaxy.cn.common.Entity;
import com.galaxy.cn.common.ObjectEntity;
import com.galaxy.cn.config.commonConfig;
import com.galaxy.cn.function.TopNHotItems;
import com.galaxy.cn.function.metricsCalculate;
import com.galaxy.cn.sink.HbaseSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.common.config.SslConfigs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.Properties;
public class Recommendation {
private static final Logger LOG = LoggerFactory.getLogger(Recommendation.class);
public static void main(String[] args) throws Exception {
//1.创建执行环境
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//指定使用事件时间
//env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
final ParameterTool parameterTool = ParameterTool.fromArgs(args);
String topic = parameterTool.get("topic", commonConfig.SOURCE_KAFKA_TOPIC);
Properties properties = parameterTool.getProperties();
properties.setProperty("group.id", commonConfig.SOURCE_KAFKA_GROUP_ID);
properties.setProperty("bootstrap.servers", commonConfig.SOURCE_KAFKA_BROKER);
properties.setProperty("session.timeout.ms", commonConfig.SESSION_TIMEOUT_MS);
properties.setProperty("max.poll.records", commonConfig.MAX_POLL_RECORD);
properties.setProperty("max.partition.fetch.bytes", commonConfig.MAX_PARTITION_FETCH_BYTES);
switch (commonConfig.KAFKA_SECURITY) {
case 1:
properties.put("security.protocol", "SSL");
properties.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
properties.put("ssl.keystore.location", commonConfig.TOOLS_LIBRARY + "keystore.jks");
properties.put("ssl.keystore.password", commonConfig.KAFKA_PIN);
properties.put("ssl.truststore.location", commonConfig.TOOLS_LIBRARY + "truststore.jks");
properties.put("ssl.truststore.password", commonConfig.KAFKA_PIN);
properties.put("ssl.key.password", commonConfig.KAFKA_PIN);
break;
case 2:
properties.put("security.protocol", "SASL_PLAINTEXT");
properties.put("sasl.mechanism", "PLAIN");
properties.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username="
+ commonConfig.KAFKA_USER + " password=" + commonConfig.KAFKA_PIN + ";");
break;
default:
}
DataStream<String> source = env.addSource(new FlinkKafkaConsumer<String>(
topic,
new SimpleStringSchema(), properties)).setParallelism(commonConfig.SOURCE_KAFKA_PARALLELISM);
WatermarkStrategy<Entity> strategy = WatermarkStrategy
.<Entity>forBoundedOutOfOrderness(Duration.ofSeconds(commonConfig.WATERMARK_TIME))
.withTimestampAssigner((Entity, timestamp) -> Entity.getCommon_recv_time()*1000);
SingleOutputStreamOperator<Entity> input = source.map(new MapFunction<String, Entity>() {
@Override
public Entity map(String message) {
Entity entity =new Entity();
try {
entity = JSON.parseObject(message, Entity.class);
if(!commonConfig.no_filter){
if(entity.getCommon_app_label()==null ||"".equals(entity.getCommon_app_label().trim())){
entity.setIfError(1);
}
}
else{
if(!commonConfig.app_white_list.containsKey(entity.getCommon_app_label())){
entity.setIfError(1);
}
}
}
catch (Exception e){
LOG.error("Entity Parsing ERROR");
entity.setIfError(1);
}
return entity;
}
}).filter(new FilterFunction<Entity>() {
@Override
public boolean filter(Entity entity) throws Exception {
return entity.ifError!=1;
}
});
KeyedStream<Entity, Tuple2<String,String>> keyedStream = input.assignTimestampsAndWatermarks(strategy).keyBy(new MyKeySelector());
SingleOutputStreamOperator<ObjectEntity> windowedStream = keyedStream
.window(SlidingEventTimeWindows.of(Time.minutes(commonConfig.SLIDINGWINDOW_TIME_MINUTE), Time.minutes(commonConfig.SLIDINGWINDOWSLOT_TIME_MINUTE)))
.process(new metricsCalculate());
SingleOutputStreamOperator<Tuple3<String,String,Long>> windoweddStream = windowedStream.keyBy(new oneKeySelector())
.process(new TopNHotItems(commonConfig.TOP_LIMIT));
windoweddStream.addSink(new HbaseSink(commonConfig.ZK_HOST)).setParallelism(commonConfig.SINK_HBASE_PARALLELISM);
env.setParallelism(commonConfig.TASK_PARALLELISM);
env.execute("RECOMMENDATION-APP-CIP");
}
public static class MyKeySelector implements KeySelector<Entity,Tuple2<String,String>> {
@Override
public Tuple2<String,String> getKey(Entity entity) throws Exception {
return new Tuple2<>(entity.getCommon_app_label(),entity.getCommon_client_ip());
}
}
public static class oneKeySelector implements KeySelector<ObjectEntity,Tuple1<String>> {
@Override
public Tuple1<String> getKey(ObjectEntity entity) throws Exception {
return new Tuple1<>(entity.getCommon_app_label());
}
}
}

74
src/main/resources/common.properties Normal file
View File

@@ -0,0 +1,74 @@
#kafka<6B>ĵ<EFBFBD>ַ<EFBFBD><D6B7>Ϣ
source.kafka.broker=192.168.44.11:9094
source.kafka.group.id =vpn-1206-1
source.kafka.topic=SESSION-RECORD-COMPLETED
source.kafka.parallelism=1
max.poll.records=3000
session.timeout.ms=60000
max.partition.fetch.bytes=31457280
#hbase<73><65>zk<7A><6B>ַ
zk.host=192.168.44.12:2181
#д<><D0B4>hbase<73><65><EFBFBD>ж<EFBFBD>
sink.hbase.parallelism=1
#д<><D0B4>hbase<73>д<EFBFBD>
sink.hbase.fm=common
#д<><D0B4>hbase<73><65><EFBFBD><EFBFBD>
sink.hbase.table=tsg_galaxy:recommendation_app_cip
#<23><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ж<EFBFBD>
task.parallelism=1
#<23><><EFBFBD><EFBFBD><EFBFBD>ӳٵȴ<D9B5>ʱ<EFBFBD>䵥λ<E4B5A5><CEBB>
watermark.time=1
#top<6F><70><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
top.limit=2
#<23><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʱ<EFBFBD>䵥λ<E4B5A5><CEBB><EFBFBD><EFBFBD>
slidingwindow.time.minute=30
#ÿ<><C3BF><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ʱ<EFBFBD>䵥λ<E4B5A5><CEBB><EFBFBD><EFBFBD>
slidingwindowslot.time.minute=1
#kafka<6B>Ƿ<EFBFBD><C7B7><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>ȫ<EFBFBD><C8AB>֤ 0<><30><EFBFBD><EFBFBD><EFBFBD><EFBFBD> 1SSL 2 SASL
kafka.security=2
#kafka SASL<53><4C>֤<EFBFBD>û<EFBFBD><C3BB><EFBFBD>
kafka.user=admin
#kafka SASL<53><4C>SSL<53><4C>֤<EFBFBD><D6A4><EFBFBD><EFBFBD>
kafka.pin=galaxy2019
#1SSL<53><4C>Ҫ
tools.library=D:\\K18-Phase2\\tsgSpace\\dat\\tsg\\
#<23>Ƿ<EFBFBD><C7B7><EFBFBD><EFBFBD><EFBFBD>ȫ<EFBFBD><C8AB>app <20>޹<EFBFBD><DEB9><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>false <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>true
has.filter=false
#ֻ<><D6BB><EFBFBD><EFBFBD>filter<65><72><EFBFBD>е<EFBFBD>common_app_label<65><6C><EFBFBD>ŷָ<C5B7> baidu.com,qq <20>ɲ<EFBFBD><C9B2><EFBFBD>д
app.white.list=
#source.kafka.broker=10.111.136.55:9092,10.111.136.56:9092,10.111.136.57:9092,10.111.136.58:9092,10.111.136.59:9092,10.111.136.60:9092,10.111.136.61:9092,10.111.136.62:9092,10.111.136.63:9092,10.111.136.64:9092,10.111.136.65:9092,10.111.136.66:9092,10.111.136.67:9092,10.111.136.68:9092,10.111.136.69:9092,10.111.136.70:9092,10.111.136.71:9092,10.111.136.72:9092,10.111.136.73:9092,10.111.136.74:9092,10.111.136.75:9092,10.111.136.76:9092,10.111.136.77:9092,10.111.136.78:9092,10.111.136.79:9092,10.111.136.80:9092,10.111.136.81:9092,10.111.136.82:9092,10.111.136.83:9092,10.111.136.84:9092,10.111.136.85:9092,10.111.136.86:9092,10.111.136.87:9092,10.111.136.88:9092,10.111.136.89:9092,10.111.136.90:9092,10.111.136.91:9092,10.111.136.92:9092,10.111.136.93:9092,10.111.136.94:9092,10.111.136.95:9092,10.111.136.96:9092,10.111.136.97:9092,10.111.136.98:9092,10.111.136.99:9092,10.111.136.100:9092,10.111.136.101:9092,10.111.136.102:9092,10.111.136.103:9092,10.111.136.104:9092
##source.kafka.broker=10.111.200.135:9092,10.111.200.136:9092,10.111.200.137:9092,10.111.200.138:9092,10.111.200.139:9092,10.111.200.140:9092,10.111.200.141:9092,10.111.200.142:9092,10.111.200.143:9092,10.111.200.144:9092,10.111.200.145:9092,10.111.200.146:9092,10.111.200.147:9092,10.111.200.148:9092,10.111.200.149:9092,10.111.200.150:9092,10.111.200.151:9092,10.111.200.152:9092,10.111.200.153:9092,10.111.200.154:9092,10.111.200.155:9092,10.111.200.156:9092,10.111.200.158:9092,10.111.200.159:9092,10.111.200.160:9092,10.111.200.161:9092,10.111.200.162:9092,10.111.200.163:9092,10.111.200.164:9092
#source.kafka.group.id=vpn-1120-1
#source.kafka.topic=test
##source.kafka.topic=CONNECTION-RECORD-COMPLETED-LOG
#source.kafka.parallelism=60
#sink.hbase.parallelism=60
#sink.hbase.fm=common
#sink.hbase.table=tsg_galaxy:recommendation_app_cip
#
#task.parallelism=60
#watermark.time=30
#top.limit=10000
#zk.host=10.111.200.165,10.111.200.166,10.111.200.167,10.111.200.168,10.111.200.169
#zk.port=2181
#slidingwindow.time.minute=30
#slidingwindowslot.time.minute=1
#
#
#
#
#kafka.security=0
##kafka SASL<53><4C>֤<EFBFBD>û<EFBFBD><C3BB><EFBFBD>
#kafka.user=admin
##kafka SASL<53><4C>SSL<53><4C>֤<EFBFBD><D6A4><EFBFBD><EFBFBD>
#kafka.pin=galaxy2019
##1SSL<53><4C>Ҫ
#tools.library=D:\\K18-Phase2\\tsgSpace\\dat\\tsg\\
#
#
##<23>Ƿ<EFBFBD><C7B7><EFBFBD><EFBFBD><EFBFBD>ȫ<EFBFBD><C8AB>app <20>޹<EFBFBD><DEB9><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>false <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>true
#has.filter=false
##ֻ<><D6BB><EFBFBD><EFBFBD>filter<65><72><EFBFBD>е<EFBFBD>common_app_label<65><6C><EFBFBD>ŷָ<C5B7> baidu.com,qq
#app.white.list=,
#

7
src/test/java/com/galaxy/cn/catalog/CatalogTest.java Normal file
View File

@@ -0,0 +1,7 @@
package com.galaxy.cn.catalog;
public class CatalogTest {
public static void main(String[] args) {
}
}