Skip to content

GitLab

B
bigdata-flink-dimensions
Commit fa3ad06e authored by caocao's avatar caocao
Browse files
Options

提交flink作业维表作业项目

parent 95c48709
Expand all Show whitespace changes
Inline Side-by-side
Showing with 890 additions and 0 deletions
.gitignore 0 → 100644
View file @ fa3ad06e
#Compile
*.class
#Mac OS
.DS_Store
#Package
*.jar
*.war
*.ear
#IntelliJ IDEA
*.idea
*.iml
#Eclipse
.classpath
.project
target/
.settings
#Log4j
logs
c2c-item-dimensions/pom.xml 0 → 100644
View file @ fa3ad06e
<?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>bigdata-flink-dimensions</artifactId>
<groupId>org.example</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>c2c-item-dimensions</artifactId>
<dependencies>
<dependency>
<groupId>com.wwdz.bigdata</groupId>
<artifactId>flink-common</artifactId>
<version>${flink.common}</version>
</dependency>
</dependencies>
<build>
<plugins>
<!-- We use the maven-shade plugin to create a fat jar that contains all necessary dependencies. -->
<!-- Change the value of <mainClass>...</mainClass> if your program entry point changes. -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.0.0</version>
<executions>
<!-- Run shade goal on package phase -->
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<artifactSet>
<excludes>
<exclude>org.apache.flink:force-shading</exclude>
<exclude>com.google.code.findbugs:jsr305</exclude>
<exclude>org.slf4j:*</exclude>
<exclude>log4j:*</exclude>
</excludes>
</artifactSet>
<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/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
<transformers>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.wwdz.bigdata.C2cShowAnalyzePortraitJob</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
<!-- Java Compiler -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<source>1.8</source>
<target>1.8</target>
</configuration>
</plugin>
<!-- Scala Compiler -->
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.2.2</version>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
<!-- Eclipse Scala Integration -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-eclipse-plugin</artifactId>
<version>2.8</version>
<configuration>
<downloadSources>true</downloadSources>
<projectnatures>
<projectnature>org.scala-ide.sdt.core.scalanature</projectnature>
<projectnature>org.eclipse.jdt.core.javanature</projectnature>
</projectnatures>
<buildcommands>
<buildcommand>org.scala-ide.sdt.core.scalabuilder</buildcommand>
</buildcommands>
<classpathContainers>
<classpathContainer>org.scala-ide.sdt.launching.SCALA_CONTAINER</classpathContainer>
<classpathContainer>org.eclipse.jdt.launching.JRE_CONTAINER</classpathContainer>
</classpathContainers>
<excludes>
<exclude>org.scala-lang:scala-library</exclude>
<exclude>org.scala-lang:scala-compiler</exclude>
</excludes>
<sourceIncludes>
<sourceInclude>**/*.scala</sourceInclude>
<sourceInclude>**/*.java</sourceInclude>
</sourceIncludes>
</configuration>
</plugin>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>build-helper-maven-plugin</artifactId>
<version>1.7</version>
<executions>
<!-- Add src/main/scala to eclipse build path -->
<execution>
<id>add-source</id>
<phase>generate-sources</phase>
<goals>
<goal>add-source</goal>
</goals>
<configuration>
<sources>
<source>src/main/scala</source>
</sources>
</configuration>
</execution>
<!-- Add src/test/scala to eclipse build path -->
<execution>
<id>add-test-source</id>
<phase>generate-test-sources</phase>
<goals>
<goal>add-test-source</goal>
</goals>
<configuration>
<sources>
<source>src/test/scala</source>
</sources>
</configuration>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-antrun-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
c2c-item-dimensions/src/main/java/com/wwdz/steam/C2cItemDimFlinkJob.java 0 → 100644
View file @ fa3ad06e
package com.wwdz.steam;
import com.wwdz.bigdata.flink.datasync.DataSyncRedisMapper;
import com.wwdz.bigdata.flink.datasync.entity.DataSyncEntity;
import com.wwdz.bigdata.flink.job.BaseFlinkJob;
import com.wwdz.bigdata.flink.job.SimpleMetricsFlinkJob;
import com.wwdz.bigdata.flink.metrics.AbstractSimpleMetricsProcessFunction;
import com.wwdz.bigdata.flink.redis.FlinkJedisConfigBaseBuilder;
import com.wwdz.bigdata.flink.source.config.DatahubConfig;
import com.wwdz.steam.pojo.ItemVo;
import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.util.Collector;
/**
* item userid维表
* 启动命令：flink run -m yarn-cluster -ynm c2c-dim-item-user -yqu common -ys 3 -ytm 2048 -yD taskmanager.memory.size=200m -d -c com.wwdz.steam.C2cItemDimFlinkJob c2c-item-dimensions-1.0-SNAPSHOT.jar
*/
public class C2cItemDimFlinkJob extends BaseFlinkJob {
public static void main(String[] args) {
StreamExecutionEnvironment env = initEnv();
String jobName = "c2cItemDim";
//
// JdbcConfigBasePojo mysqlJdbcConfig = new JdbcConfigBasePojo();
// mysqlJdbcConfig.setJdbcUrl("jdbc:mysql://drdshbgap651vvw3.drds.aliyuncs.com:3306/wwdz_item?useUnicode=true&characterEncoding=utf-8&useSSL=false");
// mysqlJdbcConfig.setDriver("com.mysql.cj.jdbc.Driver");
// mysqlJdbcConfig.setUsername("wwdz_item_read");
// mysqlJdbcConfig.setPassword("fJAXpQNOI0HVaoaz");
//// userId, activityId, created
// mysqlJdbcConfig.setUseSql("SELECT userId as user_id ,type as item_type,itemId as item_id FROM item where type =12");
// mysqlJdbcConfig.setBatchSize(500000);
//
// JdbcMysqlSource jdbcMysqlSource = new JdbcMysqlSource(mysqlJdbcConfig);
// SingleOutputStreamOperator<ItemVo> ds = env.addSource(jdbcMysqlSource).disableChaining();
DatahubConfig datahubConfig = new DatahubConfig(jobName, 274);
SimpleMetricsFlinkJob<ItemVo, DataSyncEntity> itemFlinkJob =
new SimpleMetricsFlinkJob<>(TypeInformation.of(ItemVo.class), TypeInformation.of(DataSyncEntity.class));
DataStream<DataSyncEntity> itemSourceDs = itemFlinkJob.process(env, datahubConfig,
new AbstractSimpleMetricsProcessFunction<ItemVo, DataSyncEntity>(TypeInformation.of(ItemVo.class)) {
@Override
public void processElement0(ItemVo itemVo, Collector<DataSyncEntity> out) {
Long itemType = itemVo.getType();
if (itemType != null && 12 == itemType) {
DataSyncEntity dataSyncEntity = new DataSyncEntity();
dataSyncEntity.setField("userId");
dataSyncEntity.setKey(itemVo.getItemId() + "");
dataSyncEntity.setValue(itemVo.getUserId() + "");
out.collect(dataSyncEntity);
}
}
});
// DataStream<DataSyncEntity> itemSourceDs = ds.map(new MapFunction<ItemVo, DataSyncEntity>() {
// @Override
// public DataSyncEntity map(ItemVo itemVo) throws Exception {
// DataSyncEntity dataSyncEntity = new DataSyncEntity();
// dataSyncEntity.setField("userId");
// dataSyncEntity.setKey(itemVo.getItemId() + "");
// dataSyncEntity.setValue(itemVo.getUserId() + "");
// return dataSyncEntity;
// }
// });
// itemSourceDs.print();
DataSyncRedisMapper dataSyncRedisMapper = new DataSyncRedisMapper("C2C:ITEMID", -1);
itemSourceDs.addSink(new RedisSink<DataSyncEntity>(FlinkJedisConfigBaseBuilder.buildJedisConfig("dim", 5), dataSyncRedisMapper));
try {
JobExecutionResult execute = env.execute(jobName);
} catch (Exception e) {
e.printStackTrace();
}
}
}
c2c-item-dimensions/src/main/java/com/wwdz/steam/mappper/BatchRedisMapping.java 0 → 100644
View file @ fa3ad06e
package com.wwdz.steam.mappper;
import com.wwdz.steam.pojo.ItemDo;
import org.apache.flink.streaming.connectors.redis.common.container.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.container.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
public class BatchRedisMapping implements RedisMapper<ItemDo> {
private final static String PREFIX = "DIMTABLE:VIDEO:VIDEOID_";
@Override
public RedisCommandDescription getCommandDescription() {
return new RedisCommandDescription(RedisCommand.HSET);
}
@Override
public String getKeyFromData(ItemDo data) {
return PREFIX + data.getItemId();
}
@Override
public String getValueFromData(ItemDo data) {
return data.getUserId();
}
@Override
public String getSecondKey(ItemDo data) {
return "userId";
}
@Override
public Integer getExpireSeconds(ItemDo data) {
return 60 * 60;
}
@Override
public String getPrefix() {
return PREFIX;
}
}
c2c-item-dimensions/src/main/java/com/wwdz/steam/pojo/ItemDo.java 0 → 100644
View file @ fa3ad06e
package com.wwdz.steam.pojo;
import lombok.Data;
@Data
public class ItemDo {
private String userId, itemId;
}
c2c-item-dimensions/src/main/java/com/wwdz/steam/pojo/ItemVo.java 0 → 100644
View file @ fa3ad06e
package com.wwdz.steam.pojo;
import lombok.Data;
@Data
public class ItemVo {
private Long userId, itemId, type;
}
c2c-item-dimensions/src/main/java/com/wwdz/steam/source/JdbcMysqlSource.java 0 → 100644
View file @ fa3ad06e
package com.wwdz.steam.source;
import com.wwdz.bigdata.flink.source.JdbcSourceWrapper;
import com.wwdz.bigdata.flink.source.odps.connects.pojo.JdbcConfigBasePojo;
import com.wwdz.steam.pojo.ItemVo;
public class JdbcMysqlSource extends JdbcSourceWrapper<ItemVo> {
private static final long serialVersionUID = 2944647508606760474L;
public JdbcMysqlSource(JdbcConfigBasePojo jdbcConfigBasePojo) {
super(jdbcConfigBasePojo);
}
public JdbcMysqlSource(String dataSourceId, String userSql) {
super(dataSourceId, userSql);
}
}
c2c-item-dimensions/src/main/resources/env/online/logback.xml 0 → 100644
View file @ fa3ad06e
<configuration>
<appender name="file" class="ch.qos.logback.core.FileAppender">
<file>${log.file}</file>
<append>false</append>
<encoder>
<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} ------ %msg%n</pattern>
</encoder>
</appender>
<appender name="stash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
<destination>10.100.9.154:5000</destination>
<encoder class="net.logstash.logback.encoder.LogstashEncoder">
<customFields>{"app_name":"index-portrait"}</customFields>
<throwableConverter class="net.logstash.logback.stacktrace.ShortenedThrowableConverter">
<rootCauseFirst>true</rootCauseFirst>
</throwableConverter>
</encoder>
<keepAliveDuration>5 minutes</keepAliveDuration>
</appender>
<!-- This affects logging for both user code and Flink -->
<root level="INFO">
<appender-ref ref="file"/>
<appender-ref ref="stash"/>
</root>
<!-- Uncomment this if you want to only change Flink's logging -->
<!--<logger name="org.apache.flink" level="INFO">-->
<!--<appender-ref ref="file"/>-->
<!--</logger>-->
<!-- The following lines keep the log level of common libraries/connectors on
log level INFO. The root logger does not override this. You have to manually
change the log levels here. -->
<logger name="akka" level="INFO">
<appender-ref ref="file"/>
</logger>
<logger name="org.apache.kafka" level="INFO">
<appender-ref ref="file"/>
</logger>
<logger name="org.apache.hadoop" level="INFO">
<appender-ref ref="file"/>
</logger>
<logger name="org.apache.zookeeper" level="INFO">
<appender-ref ref="file"/>
</logger>
<!-- Suppress the irrelevant (wrong) warnings from the Netty channel handler -->
<logger name="org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline" level="ERROR">
<appender-ref ref="file"/>
</logger>
</configuration>
\ No newline at end of file
c2c-item-dimensions/src/main/resources/env/online/properties.properties 0 → 100644
View file @ fa3ad06e
##item
flink.c2c.mysql.item.url=jdbc:mysql://drdshbgap651vvw3.drds.aliyuncs.com:3306/wwdz_item?useUnicode=true&characterEncoding=utf-8&useSSL=false
flink.c2c.mysql.item.driver=com.mysql.cj.jdbc.Driver
flink.c2c.mysql.item.username=wwdz_item_read
flink.c2c.mysql.item.password=fJAXpQNOI0HVaoaz
c2c-item-dimensions/src/main/resources/env/test/logback.xml 0 → 100644
View file @ fa3ad06e
<configuration>
<appender name="file" class="ch.qos.logback.core.FileAppender">
<file>${log.file}</file>
<append>false</append>
<encoder>
<pattern>%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{60} %X{sourceThread} ------ %msg%n</pattern>
</encoder>
</appender>
<appender name="stash" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
<destination>10.100.9.154:5000</destination>
<encoder class="net.logstash.logback.encoder.LogstashEncoder">
<customFields>{"app_name":"index-portrait"}</customFields>
<throwableConverter class="net.logstash.logback.stacktrace.ShortenedThrowableConverter">
<rootCauseFirst>true</rootCauseFirst>
</throwableConverter>
</encoder>
<keepAliveDuration>5 minutes</keepAliveDuration>
</appender>
<!-- This affects logging for both user code and Flink -->
<root level="INFO">
<appender-ref ref="file"/>
<appender-ref ref="stash"/>
</root>
<!-- Uncomment this if you want to only change Flink's logging -->
<!--<logger name="org.apache.flink" level="INFO">-->
<!--<appender-ref ref="file"/>-->
<!--</logger>-->
<!-- The following lines keep the log level of common libraries/connectors on
log level INFO. The root logger does not override this. You have to manually
change the log levels here. -->
<logger name="akka" level="INFO">
<appender-ref ref="file"/>
</logger>
<logger name="org.apache.kafka" level="INFO">
<appender-ref ref="file"/>
</logger>
<logger name="org.apache.hadoop" level="INFO">
<appender-ref ref="file"/>
</logger>
<logger name="org.apache.zookeeper" level="INFO">
<appender-ref ref="file"/>
</logger>
<!-- Suppress the irrelevant (wrong) warnings from the Netty channel handler -->
<logger name="org.apache.flink.shaded.akka.org.jboss.netty.channel.DefaultChannelPipeline" level="ERROR">
<appender-ref ref="file"/>
</logger>
</configuration>
\ No newline at end of file
c2c-item-dimensions/src/main/resources/env/test/properties.properties 0 → 100644
View file @ fa3ad06e
##item
flink.c2c.mysql.item.url=jdbc:mysql://rm-bp1eg0du2i8k1k039.mysql.rds.aliyuncs.com:3306/mall?useUnicode=true&characterEncoding=utf-8&useSSL=false
flink.c2c.mysql.item.username=root
flink.c2c.mysql.item.password=Zhuangdian!@#
pom.xml 0 → 100644
View file @ fa3ad06e
This diff is collapsed.
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment