msb-public
/
austin
mirror of https://github.com/ZhongFuCheng3y/austin
5
3
Code Issues Projects Releases Wiki Activity

数据仓库模块 init (未完成)

Browse Source
pull/26/head
3y 2 years ago
parent 7382e391af
commit 89c50b5bf1
11 changed files with 395 additions and 103 deletions
Show Stats Download Patch File Download Diff File

31
austin-client/pom.xml
Unescape View File

@ -1,31 +0,0 @@
<?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>austin</artifactId>
<groupId>com.java3y.austin</groupId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>austin-client</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>com.java3y.austin</groupId>
<artifactId>austin-service-api</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
</dependency>
</dependencies>
</project>

63
austin-client/src/main/java/com/java3y/austin/client/service/AustinService.java
Unescape View File

@ -1,63 +0,0 @@
package com.java3y.austin.client.service;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSON;
import cn.hutool.json.JSONUtil;
import com.java3y.austin.service.api.domain.MessageParam;
import com.java3y.austin.service.api.domain.SendRequest;
import com.java3y.austin.service.api.domain.SendResponse;
import com.java3y.austin.service.api.enums.BusinessCode;
/**
* 对外提供的接口
*
* @author 3y
*/
public class AustinService {
private static final String SEND_PATH = "/send";
private static final String RECALL_PATH = "/messageTemplate/recall";
/**
* 发送消息
*
* @param host 调用的接口host
* @param sendRequest 发送消息入参
* @return
* @throws Exception
*/
public static SendResponse send(String host, SendRequest sendRequest) throws Exception {
String url = host + SEND_PATH;
String result = HttpUtil.post(url, JSONUtil.toJsonStr(sendRequest));
return JSONUtil.toBean(result, SendResponse.class);
}
/**
* 根据模板ID撤回消息
*
* @param host 调用的接口host
* @param id 撤回消息的模板ID
* @return
* @throws Exception
*/
public static SendResponse recall(String host, String id) throws Exception {
String url = host + RECALL_PATH + StrUtil.SLASH + id;
String result = HttpUtil.post(url, id);
return JSONUtil.toBean(result, SendResponse.class);
}
public static void main(String[] args) {
SendRequest request = SendRequest.builder().code(BusinessCode.COMMON_SEND.getCode())
.messageTemplateId(68L)
.messageParam(MessageParam.builder().receiver("phone").build()).build();
try {
AustinService.send("url", request);
} catch (Exception e) {
System.out.println(e);
}
}
}

127
austin-data-house/pom.xml
Unescape View File

@ -0,0 +1,127 @@
<?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>austin</artifactId>
<groupId>com.java3y.austin</groupId>
<version>0.0.1-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>austin-data-house</artifactId>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<!-- Flink Dependency -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-hive_2.12</artifactId>
<version>1.14.3</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_2.12</artifactId>
<version>1.14.3</version>
</dependency>
<!-- Hive Dependency -->
<dependency>
<groupId>org.apache.hive</groupId>
<artifactId>hive-exec</artifactId>
<version>3.1.2</version>
</dependency>
<!--hadoop start-->
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-core</artifactId>
<version>3.3.1</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-common</artifactId>
<version>3.3.1</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-common</artifactId>
<version>3.3.1</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-mapreduce-client-jobclient</artifactId>
<version>3.3.1</version>
</dependency>
<!--hadoop end-->
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<configuration>
<createDependencyReducedPom>false</createDependencyReducedPom>
</configuration>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<filters>
<filter>
<!-- 不要拷贝 META-INF 目录下的签名，否则会引起 SecurityExceptions 。 -->
<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.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.ServicesResourceTransformer" />
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.java3y.austin.stream.AustinBootStrap</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>2.2.2.RELEASE</version>
</dependency>
</dependencies>
</plugin>
</plugins>
</build>
</project>

82
austin-data-house/src/main/java/com/java3y/austin/datahouse/AustinBootStrap.java
Unescape View File

@ -0,0 +1,82 @@
package com.java3y.austin.datahouse;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;
import java.time.Duration;
/**
* flink启动类
*
* @author 3y
*/
public class AustinBootStrap {
public static void main(String[] args) throws Exception {
TableEnvironment tableEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode());
tableEnv.getConfig().getConfiguration().set(ExecutionCheckpointingOptions.CHECKPOINTING_MODE, CheckpointingMode.EXACTLY_ONCE);
tableEnv.getConfig().getConfiguration().set(ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, Duration.ofSeconds(20));
tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
String catalogName = "my_hive";
HiveCatalog catalog = new HiveCatalog(catalogName, "austin_hive", null);
tableEnv.registerCatalog(catalogName, catalog);
tableEnv.useCatalog(catalogName);
tableEnv.executeSql("CREATE DATABASE IF NOT EXISTS austin");
tableEnv.executeSql("DROP TABLE IF EXISTS austin.message_anchor_info");
tableEnv.executeSql("create table austin.message_anchor_info(" +
"ids String,\n" +
"state String,\n" +
"businessId String,\n" +
"log_ts Timestamp(3),\n" +
"WATERMARK FOR log_ts AS log_ts -INTERVAL '5' SECOND" +
")WITH(" +
" 'connector' = 'kafka',\n" +
"'topic' = 'austinTraceLog',\n" +
" 'properties.bootstrap.servers' = 'kafka_ip:9092',\n" +
"'properties.group.id' = 'flink1',\n" +
"'scan.startup.mode' = 'earliest-offset',\n" +
"'format' = 'json',\n" +
"'json.fail-on-missing-field' = 'false',\n" +
"'json.ignore-parse-errors' = 'true'" +
")");
// tableEnv.executeSql("CREATE DATABASE IF NOT EXISTS hive_tmp");
// tableEnv.executeSql("DROP TABLE IF EXISTS hive_tmp.log_hive");
//
// tableEnv.executeSql(" CREATE TABLE hive_tmp.log_hive (\n" +
// " user_id STRING,\n" +
// " order_amount DOUBLE\n" +
// " ) PARTITIONED BY (\n" +
// " dt STRING,\n" +
// " hr STRING\n" +
// " ) STORED AS PARQUET\n" +
// " TBLPROPERTIES (\n" +
// " 'sink.partition-commit.trigger' = 'partition-time',\n" +
// " 'sink.partition-commit.delay' = '1 min',\n" +
// " 'format' = 'json',\n" +
// " 'sink.partition-commit.policy.kind' = 'metastore,success-file',\n" +
// " 'partition.time-extractor.timestamp-pattern'='$dt $hr:00:00'" +
// " )");
// tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
//
// tableEnv.executeSql("" +
// " INSERT INTO hive_tmp.log_hive\n" +
// " SELECT\n" +
// " user_id,\n" +
// " order_amount,\n" +
// " DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH')\n" +
// " FROM austin.message_anchor_info");
}
}

12
austin-data-house/src/main/java/com/java3y/austin/datahouse/constants/DataHouseConstant.java
Unescape View File

@ -0,0 +1,12 @@
package com.java3y.austin.datahouse.constants;
/**
* 数据仓库常量
*
* @author 3y
*/
public class DataHouseConstant {
public static final String HIVE_CONF_ = "austinLogGroup";
}

71
austin-data-house/src/main/resources/hive-site.xml
Unescape View File

@ -0,0 +1,71 @@
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
<!--指定mysql数据库连接的database-->
<property>
<name>javax.jdo.option.ConnectionURL</name>
<value>jdbc:postgresql://3y_ip:5432/metastore?createDatabaseIfNotExist=true</value>
<description>JDBC connect string for a JDBC metastore</description>
</property>
<property>
<name>javax.jdo.option.ConnectionDriverName</name>
<value>org.postgresql.Driver</value>
<description>Driver class name for a JDBC metastore</description>
</property>
<property>
<name>javax.jdo.option.ConnectionUserName</name>
<value>hive</value>
<description>username to use against metastore database</description>
</property>
<property>
<name>javax.jdo.option.ConnectionPassword</name>
<value>hive</value>
<description>password to use against metastore database</description>
</property>
<property>
<name>hive.metastore.uris</name>
<value>thrift://3y_ip:9083</value>
<description>Thrift URI for the remote metastore. Used by metastore client to connect to remote metastore.
</description>
</property>
<property>
<name>datanucleus.schema.autoCreateAll</name>
<value>true</value>
</property>
<property>
<name>hive.server2.logging.operation.log.location</name>
<value>/root/sd/apache-hive-2.3.4-bin/tmp/operation_logs</value>
<description>Top level directory where operation logs are stored if logging functionality is enabled
</description>
</property>
<property>
<name>hive.exec.scratchdir</name>
<value>/root/sd/apache-hive-2.3.4-bin/tmp/hive</value>
<description>HDFS root scratch dir for Hive jobs which gets created with write all (733) permission. For each
connecting
user, an HDFS scratch dir: ${hive.exec.scratchdir}/&lt;username&gt; is created, with
${hive.scratch.dir.permission}.
</description>
</property>
<property>
<name>hive.exec.local.scratchdir</name>
<value>/root/sd/apache-hive-2.3.4-bin/tmp/hive/local</value>
<description>Local scratch space for Hive jobs</description>
</property>
<property>
<name>hive.downloaded.resources.dir</name>
<value>/root/sd/apache-hive-2.3.4-bin/tmp/hive/resources</value>
<description>Temporary local directory for added resources in the remote file system.</description>
</property>
</configuration>

5
austin-data-house/src/main/resources/sql-client-defaults.yaml
Unescape View File

@ -0,0 +1,5 @@
catalogs:
- name: my_hive
type: hive
default-database: austin_hive
hive-conf-dir: /opt/flink/conf

29
docker/flink/docker-compose.yml
Unescape View File

@ -1,27 +1,40 @@
version: "2.2"
services:
jobmanager:
image: flink:latest
image: flink:1.14.3
ports:
- "8081:8081"
command: jobmanager
volumes:
- ./sql-client-defaults.yaml:/opt/flink/conf/sql-client-defaults.yaml:rw
- ./hive-site.xml:/opt/flink/conf/hive-site.xml:rw
environment:
- |
FLINK_PROPERTIES=
jobmanager.rpc.address: jobmanager
- SET_CONTAINER_TIMEZONE=true
- CONTAINER_TIMEZONE=Asia/Shanghai
- TZ=Asia/Shanghai
taskmanager:
image: flink:latest
image: flink:1.14.3
depends_on:
- jobmanager
command: taskmanager
volumes:
- ./sql-client-defaults.yaml:/opt/flink/conf/sql-client-defaults.yaml:rw
- ./hive-site.xml:/opt/flink/conf/hive-site.xml:rw
environment:
- |
FLINK_PROPERTIES=
jobmanager.rpc.address: jobmanager
taskmanager.numberOfTaskSlots: 2
- SET_CONTAINER_TIMEZONE=true
- CONTAINER_TIMEZONE=Asia/Shanghai
- TZ=Asia/Shanghai
sql-client:
image: flink:1.14.3
volumes:
- ./sql-client-defaults.yaml:/opt/flink/conf/sql-client-defaults.yaml:rw
- ./hive-site.xml:/opt/flink/conf/hive-site.xml:rw
command: ./bin/sql-client.sh embedded ../conf/sql-client-defaults.yaml
depends_on:
- jobmanager
environment:
- |
FLINK_PROPERTIES=
jobmanager.rpc.address: jobmanager
rest.address: jobmanager

71
docker/flink/hive-site.xml
Unescape View File

@ -0,0 +1,71 @@
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
<!--指定mysql数据库连接的database-->
<property>
<name>javax.jdo.option.ConnectionURL</name>
<value>jdbc:postgresql://3y_ip:5432/metastore?createDatabaseIfNotExist=true</value>
<description>JDBC connect string for a JDBC metastore</description>
</property>
<property>
<name>javax.jdo.option.ConnectionDriverName</name>
<value>org.postgresql.Driver</value>
<description>Driver class name for a JDBC metastore</description>
</property>
<property>
<name>javax.jdo.option.ConnectionUserName</name>
<value>hive</value>
<description>username to use against metastore database</description>
</property>
<property>
<name>javax.jdo.option.ConnectionPassword</name>
<value>hive</value>
<description>password to use against metastore database</description>
</property>
<property>
<name>hive.metastore.uris</name>
<value>thrift://3y_ip:9083</value>
<description>Thrift URI for the remote metastore. Used by metastore client to connect to remote metastore.
</description>
</property>
<property>
<name>datanucleus.schema.autoCreateAll</name>
<value>true</value>
</property>
<property>
<name>hive.server2.logging.operation.log.location</name>
<value>/root/sd/apache-hive-2.3.4-bin/tmp/operation_logs</value>
<description>Top level directory where operation logs are stored if logging functionality is enabled
</description>
</property>
<property>
<name>hive.exec.scratchdir</name>
<value>/root/sd/apache-hive-2.3.4-bin/tmp/hive</value>
<description>HDFS root scratch dir for Hive jobs which gets created with write all (733) permission. For each
connecting
user, an HDFS scratch dir: ${hive.exec.scratchdir}/&lt;username&gt; is created, with
${hive.scratch.dir.permission}.
</description>
</property>
<property>
<name>hive.exec.local.scratchdir</name>
<value>/root/sd/apache-hive-2.3.4-bin/tmp/hive/local</value>
<description>Local scratch space for Hive jobs</description>
</property>
<property>
<name>hive.downloaded.resources.dir</name>
<value>/root/sd/apache-hive-2.3.4-bin/tmp/hive/resources</value>
<description>Temporary local directory for added resources in the remote file system.</description>
</property>
</configuration>

5
docker/flink/sql-client-defaults.yaml
Unescape View File

@ -0,0 +1,5 @@
catalogs:
- name: default_catalog
type: hive
default-database: austin_hive
hive-conf-dir: /opt/flink/conf

2
pom.xml
Unescape View File

@ -13,7 +13,7 @@
<module>austin-handler</module>
<module>austin-cron</module>
<module>austin-stream</module>
<module>austin-client</module>
<module>austin-data-house</module>
</modules>
<parent>

Write Preview
Loading…
Cancel
Save