case "$1" in start) if [ -f ${PIDFILE} ];then echo "${PIDFILE} exists, process is already running or crashed" else echo "Starting Redis Server ...." ${EXEC} ${CONF} fi if [ "$?" = "0" ];then echo "Redis is running" fi ;; stop) if [ ! -e ${PIDFILE} ];then echo "${PIDFILE} does not exists, process is not running" else PID=$(cat ${PIDFILE}) echo "Stopping..." ${REDIS_CLI} -p ${REDIS_PORT} AUTH root ${REDIS_CLI} -p ${REDIS_PORT} SHUTDOWN while [ -x ${PIDFILE} ] do echo "Waiting for Redis shutdown..." sleep 1 done echo "Redis stoped" fi ;; restart|force-reload) ${0} stop ${0} start ;; *) echo "Usage: /etc/init.d/redis {start|stop|restart|force-reload}" >&2 exit 1 esac
##9. 提交jar [root@hadoop flink-1.9.1]# flink run examples/batch/WordCount.jar --input /etc/profile --output /home/output/00 Starting execution of program Program execution finished Job with JobID ca414928ddcfd57969326827f5cb7205 has finished. Job Runtime: 21670 ms
[root@hadoop flink-1.9.1]# tail -f /home/output/00 want 1 we 1 what 1 wide 1 will 1 workaround 1 x 1 you 3 your 1 z 1
package com.qf.bigdata.realtime.constant;
import java.io.Serializable;
import java.time.format.DateTimeFormatter;
import java.time.temporal.ChronoUnit;
public class CommonConstant implements Serializable {
public static final int DEF_NUMBER_ZERO = 0;
public static final int DEF_NUMBER_ONE = 1;
public static final int DEF_NUMBER_DUL = 2;
//用户数量限制级别
public static final Integer USER_COUNT_LEVEL = 5;
public static final int DEF_CODE_COUNT = 4; //代码位数
public static final int DEF_RANGER = 10; //范围
//时间格式
public static final DateTimeFormatter PATTERN_YYYYMMDD = DateTimeFormatter.ofPattern("yyyyMMdd");
public static final DateTimeFormatter PATTERN_YYYYMMDD_MID = DateTimeFormatter.ofPattern("yyyy-MM-dd");
public static final DateTimeFormatter PATTERN_HOUR = DateTimeFormatter.ofPattern("HH");
public static final String FORMATTER_YYYYMMDD = "yyyyMMdd";
public static final String FORMATTER_YYYYMMDD_MID = "yyyy-MM-dd";
public static final String FORMATTER_YYYYMMDDHHMMDD = "yyyyMMddHHmmss";
public static final String FORMATTER_YYYYMMDDHHMM = "yyyyMMddHHmm";
public static final String FORMATTER_YYYYMMDDHH = "yyyyMMddHH";
//发送序列化对象
public static final ChronoUnit chronoUnit = ChronoUnit.MINUTES;
public static final ChronoUnit dayChronoUnit = ChronoUnit.DAYS;
//===charset====================================================================
public static final String CHARSET_UTF8 = "utf-8"; //测试通道
//===kafka-topic====================================================================
public static final String TOPIC_TEST = "t-release"; //投放topic
public static final String KAFKA_PRODUCER_JSON_PATH = "kafka/json/kafka-producer.properties";
public static final String KAFKA_CONSUMER_JSON_PATH = "kafka/json/kafka-consumer.properties";
public static final String KAFKA_CONFIG_PATH = "kafka/kafka-config.properties";
//===zk====================================================================
public static final String ZK_CONNECT = "zk.connect";
public static final String ZK_CONNECT_KAFKA = "zk.kafka.connect";
public static final String ZK_SESSION_TIMEOUT = "zk.session.timeout";
public static final String ZK_CONN_TIMEOUT = "zk.connection.timeout";
public static final String ZK_BEE_ROOT = "zk.dw.root";
//===常用符号====================================================================
public static final String Encoding_UTF8 = "UTF-8";
public static final String Encoding_GBK = "GBK";
public static final String MIDDLE_LINE = "-";
public static final String BOTTOM_LINE = "_";
public static final String COMMA = ",";
public static final String SEMICOLON = ";";
public static final String PLINE = "|";
public static final String COLON = ":";
public static final String PATH_W = "\\";
public static final String PATH_L = "/";
public static final String POINT = ".";
public static final String BLANK = " ";
public static final String LEFT_ARROWS = "<-";
public static final String RIGHT_ARROWS = "->";
public static final String LEFT_BRACKET = "[";
public static final String RIGHT_BRACKET = "]";
public static final String TAB = "\t";
//=====================================================
public static final String KAFKA_DATA_KEY_TOPIC = "topic";
public static final String KEY_TIMESTAMP = "timestamp";
public static final String KEY_KEY = "key";
public static final String KEY_VALUE = "value";
public static final String KEY_OFFSET = "offset";
public static final String KEY_PARTITION = "partition";
public static final String KEY_CTTIME_BEGIN = "ctTimeBegin";
public static final String KEY_CTTIME_END = "ctTimeEnd";
public static final String KEY_CTTIME = "ctTime";
public static final String KEY_USER_CODE = "userCode";
public static final String KEY_ORDER_CODE = "orderCode";
public static final String KEY_VEHICLE_CODE = "vehicleCode";
public static final String KEY_VEHICLE_TYPE = "vehicleType";
public static final String KEY_STATUS = "status";
public static final String KEY_LNG = "longitude";
public static final String KEY_LAT = "latitude";
public static final String KEY_GEOHASH = "geoHash";
public static final String KEY_ADCODE = "adcode";
public static final String KEY_PROVINCE = "province";
public static final String KEY_DISTRICT = "district";
public static final String KEY_TOWNCODE = "towncode";
public static final String KEY_TOWNSHIP = "township";
public static final String KEY_FORMATTED_ADDRESS = "formatted_address";
public static final String KEY_ADDRESS = "address";
public static final String KEY_G_SIGNAL = "gSignal";
//GIS
public final static Double LONGITUDE_CHINA_MAX = 135.05; //经度
public final static Double LONGITUDE_CHINA_MIN = 73.66; //经度
public final static Double LATITUDE_CHINA_MAX = 53.55; //纬度
public final static Double LATITUDE_CHINA_MIN = 3.86; //纬度
//多省
public final static Double LONGITUDE_REGION_MAX = 125.30; //经度
public final static Double LONGITUDE_REGION_MIN = 113.25; //经度
public final static Double LATITUDE_REGION_MAX = 34.365791; //纬度
public final static Double LATITUDE_REGION_MIN = 30.038888; //纬度
//北京行政中心的纬度为39.92，经度为116.46
//北京 北纬39”26'至41”03'，东经115”25'至117”30'
public final static Double LONGITUDE_BJ_MAX = 117.30; //经度
public final static Double LONGITUDE_BJ_MIN = 115.25; //经度
public final static Double LATITUDE_BJ_MAX = 41.03; //纬度
public final static Double LATITUDE_BJ_MIN = 39.26; //纬度
//学校附近
public final static Double LONGITUDE_QF_MIN = 116.300000; //经度
public final static Double LONGITUDE_QF_MAX = 116.399999; //经度
public final static Double LATITUDE_QF_MIN = 40.060000; //纬度
public final static Double LATITUDE_QF_MAX = 40.069999; //纬度
}
package com.qf.bigdata.realtime.dvo;
//国家地区
public class RegionDo {
public String regionCode ;
public String regionCodeDesc ;
public String regionCity;
public String reginCityDesc;
public String regionProvince;
public String regionProvinceDesc;
public String getRegionCode() {
return regionCode;
}
public String getRegionCodeDesc() {
return regionCodeDesc;
}
public String getRegionCity() {
return regionCity;
}
public String getReginCityDesc() {
return reginCityDesc;
}
public String getRegionProvince() {
return regionProvince;
}
public String getRegionProvinceDesc() {
return regionProvinceDesc;
}
public void setRegionCode(String regionCode) {
this.regionCode = regionCode;
}
public void setRegionCodeDesc(String regionCodeDesc) {
this.regionCodeDesc = regionCodeDesc;
}
public void setRegionCity(String regionCity) {
this.regionCity = regionCity;
}
public void setReginCityDesc(String reginCityDesc) {
this.reginCityDesc = reginCityDesc;
}
public void setRegionProvince(String regionProvince) {
this.regionProvince = regionProvince;
}
public void setRegionProvinceDesc(String regionProvinceDesc) {
this.regionProvinceDesc = regionProvinceDesc;
}
public void setRegionCityDesc(String orDefault) {
}
}
package com.qf.bigdata.realtime.util;
import com.qf.bigdata.realtime.dvo.RegionDo;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
// *** 作csv文件的工具类
public class CSVUtil implements Serializable {
private static final Logger log = LoggerFactory.getLogger(CSVUtil.class);
public static final String AREA_CODE_CSV_FILE = "areacode/areacode.csv";
public static final String AREA_CODE_GIS_LOCATION_CHINA_CSV_FILE = "areacode/china_gis_location.csv";
public static final String REGION_FILE = "areacode/dim_region.csv";
public static final String PUB_FILE = "areacode/dim_pub.csv";
public static final String PRODUCT_FILE = "areacode/dim_product.csv";
public static final char QUOTE_TAB = '\t';
public static final char QUOTE_COMMON = ',';
public static final char PLINE = '|';
public static final char PATH_EQ = '=';
public static final String NEW_LINE_SEPARATOR = "\n";
//读取csv文件
public static List> readCSVFile(String path,char delimiter){
//创建返回结果
List> result = new ArrayList<>();
Reader reader = null;
try{
reader = new InputStreamReader(CSVUtil.class.getClassLoader().getResourceAsStream(path));
//通过流获取csv的解析器
CSVParser csvParser = new CSVParser(
reader,
CSVFormat.DEFAULT
.withFirstRecordAsHeader() //第一行数据是作为表头，而不是数据
.withDelimiter(delimiter) //设置分隔符
.withIgnoreHeaderCase() //忽略header
.withTrim() //去掉前面和后面的空格
);
//获取所有的列名
Map header = csvParser.getHeaderMap();
Set colKeys = header.keySet();
//遍历csv的解析器，获取其他的每一个csv的记录
for(CSVRecord csdvRecord:csvParser){
//将reacord对象转换为map对象，得到每行的值
Map values = csdvRecord.toMap();
//遍历列名的列表
for(String colKey: colKeys){
String colValue = csdvRecord.get(colKey);
values.put(colKey,colValue);
}
//把value的map添加到最终结果集合中
result.add(values);
}
}catch (Exception e){
log.error("read.csvfile.error",path);
}finally {
if(null != reader){
try{
reader.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return result;
}
// public static void main(String[] args) {
// List> list = readCSVFile(REGION_FILE,',');
// System.out.println(list);
// }
}
特别说明上面最后几行注释就是测试是否可以读取csv文件，如图运行。
package com.qf.bigdata.realtime.util;
import org.apache.commons.lang3.math.NumberUtils;
import java.io.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
public class QParameterTool implements Serializable, Cloneable{
private static final long serialVersionUID = 1L;
protected static final String NO_VALUE_KEY = "__NO_VALUE_KEY";
protected static final String DEFAULT_UNDEFINED = "";
// ------------------ Constructors ------------------------
/**
* Returns {@link QParameterTool} for the given arguments. The arguments are keys followed by values.
* Keys have to start with '-' or '--'
*
*
Example arguments:
* --key1 value1 --key2 value2 -key3 value3
*
* @param args Input array arguments
* @return A {@link QParameterTool}
*/
public static QParameterTool fromArgs(String[] args) {
final Map map = new HashMap<>(args.length / 2);
int i = 0;
while (i < args.length) {
final String key;
if (args[i].startsWith("--")) {
key = args[i].substring(2);
} else if (args[i].startsWith("-")) {
key = args[i].substring(1);
} else {
throw new IllegalArgumentException(
String.format("Error parsing arguments '%s' on '%s'. Please prefix keys with -- or -.",
Arrays.toString(args), args[i]));
}
if (key.isEmpty()) {
throw new IllegalArgumentException(
"The input " + Arrays.toString(args) + " contains an empty argument");
}
i += 1; // try to find the value
if (i >= args.length) {
map.put(key, NO_VALUE_KEY);
} else if (NumberUtils.isNumber(args[i])) {
map.put(key, args[i]);
i += 1;
} else if (args[i].startsWith("--") || args[i].startsWith("-")) {
// the argument cannot be a negative number because we checked earlier
// -> the next argument is a parameter name
map.put(key, NO_VALUE_KEY);
} else {
map.put(key, args[i]);
i += 1;
}
}
return fromMap(map);
}
/**
* Returns {@link QParameterTool} for the given {@link Properties} file.
*
* @param path Path to the properties file
* @return A {@link QParameterTool}
* @throws IOException If the file does not exist
* @see Properties
*/
public static QParameterTool fromPropertiesFile(String path) throws IOException {
File propertiesFile = new File(path);
return fromPropertiesFile(propertiesFile);
}
/**
* Returns {@link QParameterTool} for the given {@link Properties} file.
*
* @param file File object to the properties file
* @return A {@link QParameterTool}
* @throws IOException If the file does not exist
* @see Properties
*/
public static QParameterTool fromPropertiesFile(File file) throws IOException {
if (!file.exists()) {
throw new FileNotFoundException("Properties file " + file.getAbsolutePath() + " does not exist");
}
try (FileInputStream fis = new FileInputStream(file)) {
return fromPropertiesFile(fis);
}
}
/**
* Returns {@link QParameterTool} for the given InputStream from {@link Properties} file.
*
* @param inputStream InputStream from the properties file
* @return A {@link QParameterTool}
* @throws IOException If the file does not exist
* @see Properties
*/
public static QParameterTool fromPropertiesFile(InputStream inputStream) throws IOException {
Properties props = new Properties();
props.load(inputStream);
return fromMap((Map) props);
}
/**
* Returns {@link QParameterTool} for the given map.
*
* @param map A map of arguments. Both Key and Value have to be Strings
* @return A {@link QParameterTool}
*/
public static QParameterTool fromMap(Map map) {
return new QParameterTool(map);
}
/**
* Returns {@link QParameterTool} from the system properties.
* Example on how to pass system properties:
* -Dkey1=value1 -Dkey2=value2
*
* @return A {@link QParameterTool}
*/
public static QParameterTool fromSystemProperties() {
return fromMap((Map) System.getProperties());
}
// ------------------ QParameterTool ------------------------
public Map data = null;
// data which is only used on the client and does not need to be transmitted
protected transient Map defaultData;
protected transient Set unrequestedParameters;
private QParameterTool(Map data) {
this.data = Collections.unmodifiableMap(new HashMap<>(data));
this.defaultData = new ConcurrentHashMap<>(data.size());
this.unrequestedParameters = Collections.newSetFromMap(new ConcurrentHashMap<>(data.size()));
unrequestedParameters.addAll(data.keySet());
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
QParameterTool that = (QParameterTool) o;
return Objects.equals(data, that.data) &&
Objects.equals(defaultData, that.defaultData) &&
Objects.equals(unrequestedParameters, that.unrequestedParameters);
}
@Override
public int hashCode() {
return Objects.hash(data, defaultData, unrequestedParameters);
}
/**
* Returns the set of parameter names which have not been requested with
* {@link #has(String)} or one of the {@code get} methods. Access to the
* map returned by {@link #toMap()} is not tracked.
*/
public Set getUnrequestedParameters() {
return Collections.unmodifiableSet(unrequestedParameters);
}
// ------------------ Get data from the util ----------------
/**
* Returns number of parameters in {@link QParameterTool}.
*/
public int getNumberOfParameters() {
return data.size();
}
/**
* Returns the String value for the given key.
* If the key does not exist it will return null.
*/
public String get(String key) {
addToDefaults(key, null);
unrequestedParameters.remove(key);
return data.get(key);
}
/**
* Returns the String value for the given key.
* If the key does not exist it will throw a {@link RuntimeException}.
*/
public String getRequired(String key) {
addToDefaults(key, null);
String value = get(key);
if (value == null) {
throw new RuntimeException("No data for required key '" + key + "'");
}
return value;
}
/**
* Returns the String value for the given key.
* If the key does not exist it will return the given default value.
*/
public String get(String key, String defaultValue) {
addToDefaults(key, defaultValue);
String value = get(key);
if (value == null) {
return defaultValue;
} else {
return value;
}
}
/**
* Check if value is set.
*/
public boolean has(String value) {
addToDefaults(value, null);
unrequestedParameters.remove(value);
return data.containsKey(value);
}
// -------------- Integer
/**
* Returns the Integer value for the given key.
* The method fails if the key does not exist or the value is not an Integer.
*/
public int getInt(String key) {
addToDefaults(key, null);
String value = getRequired(key);
return Integer.parseInt(value);
}
/**
* Returns the Integer value for the given key. If the key does not exists it will return the default value given.
* The method fails if the value is not an Integer.
*/
public int getInt(String key, int defaultValue) {
addToDefaults(key, Integer.toString(defaultValue));
String value = get(key);
if (value == null) {
return defaultValue;
}
return Integer.parseInt(value);
}
// -------------- LONG
/**
* Returns the Long value for the given key.
* The method fails if the key does not exist.
*/
public long getLong(String key) {
addToDefaults(key, null);
String value = getRequired(key);
return Long.parseLong(value);
}
/**
* Returns the Long value for the given key. If the key does not exists it will return the default value given.
* The method fails if the value is not a Long.
*/
public long getLong(String key, long defaultValue) {
addToDefaults(key, Long.toString(defaultValue));
String value = get(key);
if (value == null) {
return defaultValue;
}
return Long.parseLong(value);
}
// -------------- FLOAT
/**
* Returns the Float value for the given key.
* The method fails if the key does not exist.
*/
public float getFloat(String key) {
addToDefaults(key, null);
String value = getRequired(key);
return Float.valueOf(value);
}
/**
* Returns the Float value for the given key. If the key does not exists it will return the default value given.
* The method fails if the value is not a Float.
*/
public float getFloat(String key, float defaultValue) {
addToDefaults(key, Float.toString(defaultValue));
String value = get(key);
if (value == null) {
return defaultValue;
} else {
return Float.valueOf(value);
}
}
// -------------- DOUBLE
/**
* Returns the Double value for the given key.
* The method fails if the key does not exist.
*/
public double getDouble(String key) {
addToDefaults(key, null);
String value = getRequired(key);
return Double.valueOf(value);
}
/**
* Returns the Double value for the given key. If the key does not exists it will return the default value given.
* The method fails if the value is not a Double.
*/
public double getDouble(String key, double defaultValue) {
addToDefaults(key, Double.toString(defaultValue));
String value = get(key);
if (value == null) {
return defaultValue;
} else {
return Double.valueOf(value);
}
}
// -------------- BOOLEAN
/**
* Returns the Boolean value for the given key.
* The method fails if the key does not exist.
*/
public boolean getBoolean(String key) {
addToDefaults(key, null);
String value = getRequired(key);
return Boolean.valueOf(value);
}
/**
* Returns the Boolean value for the given key. If the key does not exists it will return the default value given.
* The method returns whether the string of the value is "true" ignoring cases.
*/
public boolean getBoolean(String key, boolean defaultValue) {
addToDefaults(key, Boolean.toString(defaultValue));
String value = get(key);
if (value == null) {
return defaultValue;
} else {
return Boolean.valueOf(value);
}
}
// -------------- SHORT
/**
* Returns the Short value for the given key.
* The method fails if the key does not exist.
*/
public short getShort(String key) {
addToDefaults(key, null);
String value = getRequired(key);
return Short.valueOf(value);
}
/**
* Returns the Short value for the given key. If the key does not exists it will return the default value given.
* The method fails if the value is not a Short.
*/
public short getShort(String key, short defaultValue) {
addToDefaults(key, Short.toString(defaultValue));
String value = get(key);
if (value == null) {
return defaultValue;
} else {
return Short.valueOf(value);
}
}
// -------------- BYTE
/**
* Returns the Byte value for the given key.
* The method fails if the key does not exist.
*/
public byte getByte(String key) {
addToDefaults(key, null);
String value = getRequired(key);
return Byte.valueOf(value);
}
/**
* Returns the Byte value for the given key. If the key does not exists it will return the default value given.
* The method fails if the value is not a Byte.
*/
public byte getByte(String key, byte defaultValue) {
addToDefaults(key, Byte.toString(defaultValue));
String value = get(key);
if (value == null) {
return defaultValue;
} else {
return Byte.valueOf(value);
}
}
// --------------- Internals
protected void addToDefaults(String key, String value) {
final String currentValue = defaultData.get(key);
if (currentValue == null) {
if (value == null) {
value = DEFAULT_UNDEFINED;
}
defaultData.put(key, value);
} else {
// there is already an entry for this key. Check if the value is the undefined
if (currentValue.equals(DEFAULT_UNDEFINED) && value != null) {
// update key with better default value
defaultData.put(key, value);
}
}
}
// ------------------------- Export to different targets -------------------------
/**
* @return A {@link Properties}
*/
public Properties getProperties() {
Properties props = new Properties();
props.putAll(this.data);
return props;
}
/**
* Create a properties file with all the known parameters (call after the last get*() call).
* Set the default value, if available.
*
*
Use this method to create a properties file skeleton.
*
* @param pathToFile Location of the default properties file.
*/
public void createPropertiesFile(String pathToFile) throws IOException {
createPropertiesFile(pathToFile, true);
}
/**
* Create a properties file with all the known parameters (call after the last get*() call).
* Set the default value, if overwrite is true.
*
* @param pathToFile Location of the default properties file.
* @param overwrite Boolean flag indicating whether or not to overwrite the file
* @throws IOException If overwrite is not allowed and the file exists
*/
public void createPropertiesFile(String pathToFile, boolean overwrite) throws IOException {
final File file = new File(pathToFile);
if (file.exists()) {
if (overwrite) {
file.delete();
} else {
throw new RuntimeException("File " + pathToFile + " exists and overwriting is not allowed");
}
}
final Properties defaultProps = new Properties();
defaultProps.putAll(this.defaultData);
try (final OutputStream out = new FileOutputStream(file)) {
defaultProps.store(out, "Default file created by Flink's ParameterUtil.createPropertiesFile()");
}
}
protected Object clone() throws CloneNotSupportedException {
return new QParameterTool(this.data);
}
// ------------------------- Interaction with other ParameterUtils -------------------------
/**
* Merges two {@link QParameterTool}.
*
* @param other Other {@link QParameterTool} object
* @return The Merged {@link QParameterTool}
*/
public QParameterTool mergeWith(QParameterTool other) {
final Map resultData = new HashMap<>(data.size() + other.data.size());
resultData.putAll(data);
resultData.putAll(other.data);
final QParameterTool ret = new QParameterTool(resultData);
final HashSet requestedParametersLeft = new HashSet<>(data.keySet());
requestedParametersLeft.removeAll(unrequestedParameters);
final HashSet requestedParametersRight = new HashSet<>(other.data.keySet());
requestedParametersRight.removeAll(other.unrequestedParameters);
ret.unrequestedParameters.removeAll(requestedParametersLeft);
ret.unrequestedParameters.removeAll(requestedParametersRight);
return ret;
}
// ------------------------- ExecutionConfig.UserConfig interface -------------------------
public Map toMap() {
return data;
}
// ------------------------- Serialization ---------------------------------------------
private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException {
in.defaultReadObject();
defaultData = new ConcurrentHashMap<>(data.size());
unrequestedParameters = Collections.newSetFromMap(new ConcurrentHashMap<>(data.size()));
}
}
package com.qf.bigdata.realtime.util;
import com.qf.bigdata.realtime.constant.CommonConstant;
import com.qf.bigdata.realtime.dvo.RegionDo;
import com.qf.bigdata.realtime.util.CSVUtil;
import com.qf.bigdata.realtime.util.QParameterTool;
import com.sun.org.apache.regexp.internal.RE;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.security.LdapGroupsMapping;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Int;
import java.io.Serializable;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class TravelOrderHelper implements Serializable {
//日志管理工具
private static final Logger log = LoggerFactory.getLogger(TravelOrderHelper.class);
/*
*订单数据属性
* 用户唯一编号
*/
public static final String KEY_KAFKA_ID = "KAFKA_ID";
public static final String KEY_ORDER_ID = "order_id";
public static final String KEY_USER_ID = "userid";
public static final String KEY_USER_MOBILE = "user_mobile";
public static final String KEY_PRODUCT_ID = "product_id";
public static final String KEY_PRODUCT_TRAFFIC = "product_traffic";
public static final String KEY_PRODUCT_TRAFFIC_GRADE = "product_traffic_grade";
public static final String KEY_PRODUCT_TRAFFIC_TYPE = "product_traffic_type";
public static final String KEY_PRODUCT_PUB = "product_pub";
public static final String KEY_USER_REGION = "user_region";
public static final String KEY_TRAVEL_MEMBER_ADULT = "travel_member_adult";
public static final String KEY_TRAVEL_MEMBER_YOUNGER = "travel_member_younger";
public static final String KEY_TRAVEL_MEMBER_BABY = "travel_member_baby";
public static final String KEY_PRODUCT_PRICE = "product_price";
public static final String KEY_HAS_ACTIVITY = "has_activity";
public static final String KEY_PRODUCT_FEE = "product_fee";
public static final String KEY_ORDER_CT = "order_ct";
// 地区
/*
* REGION_KEY_CODE : 区域码
* REGION_KEY_CODE_DESC = 区域码说明
* REGION_KEY_CITY : 区域城市
* REGION_KEY_CITY_DESC : 区域城市描述
* REGION_KEY_PROVINCE : 区域省份
* REGION_KEY_PROVINCE_DESC : 区域省份描述
*/
public static final String REGION_KEY_CODE = "region_code";
public static final String REGION_KEY_CODE_DESC = "region_code_desc";
public static final String REGION_KEY_CITY = "region_city";
public static final String REGION_KEY_CITY_DESC = "region_city_desc";
public static final String REGION_KEY_PROVINCE = "region_province";
public static final String REGION_KEY_PROVINCE_DESC = "region_province_desc";
// 地区列表，里面存放了所有的地区
public static final List regions = new ArrayList<>();
// 酒店
/*
* PUB_KEY_ID : '酒店的编号',
* PUB_KEY_NAME : '酒店的名称',
* PUB_KEY_STAT : '酒店的星级',
* PUB_KEY_GRADE : '酒店的等级编码',
* PUB_KEY_GRADE_DESC : '酒店的等级编码描述',
* PUB_KEY_AREA_CODE : '酒店所在区域编码',
* PUB_KEY_ADDRESS : '酒店地址',
* PUB_KEY_IS_NATIONAL : '是否属于境内',
*/
public static final String PUB_KEY_ID = "pub_id";
public static final String PUB_KEY_NAME = "pub_name";
public static final String PUB_KEY_STAT = "pub_stat";
public static final String PUB_KEY_GRADE = "pub_grade";
public static final String PUB_KEY_GRADE_DESC = "pub_grade_desc";
public static final String PUB_KEY_AREA_CODE = "pub_area_code";
public static final String PUB_KEY_ADDRESS = "pub_address";
public static final String PUB_KEY_IS_NATIONAL = "is_national";
// 酒店与旅游产品的映射集合
public static final Map proMappingPub = new HashMap<>();
//产品
public static final String PRODUCT_KEY_ID = "product_id";
public static final String SHOP_KEY_ID = "shop_id";
// 产品的信息列表
public static final List products = new ArrayList<>();
//参数
public static final String KEY_TOPIC = "topic";
public static final String KEY_SOURCE = "source";
public static final String KEY_BEGIN = "begin";
public static final String KEY_END = "end";
public static final String KEY_COUNT = "count";
public static final String KEY_SLEEP = "sleep";
public static final String SOURCE_PRODUCT = "product";
public static final String SOURCE_PAY = "pay";
public static final String TIME_ORDER = "time_order";
public static final Integer TIME_RANGE_MIN = 1;
public static final Integer TIME_RANGE_MAX = 120;
public static final Integer COUNT_MIN = 1;
public static final Integer COUNT_MAX = 10000;
public static final Integer SLEEP_MIN = 1;
public static final Integer SLEEP_MAX = 3600 * 1000;
/*
* 造数据（订单数据）的入口
* --topic t_travel_orders
* */
public static void main(String[] args) {
//校验这个参数是否合法
String checkResult = checkParams(args);
//校验结果
if (StringUtils.isEmpty(checkResult)){
//2.1执行if说明参数无问题，效果是连续不断地创作数据，然后将数据生产到指定的kafka主题
chooseFun(args);
}else {
System.err.println("some erros happends ["+checkResult+"]");
}
}
//校验参数是否合法
//如果这个方法返回的是空字符串，说明参数没有问题
private static String checkParams(String[] args) {
//定义一个返回结果的字符串
String result = "";
//通过加载args参数列表获取参数工具类对象
QParameterTool tools = QParameterTool.fromArgs(args);
//校验主题
String topic = tools.get(KEY_TOPIC);
if(StringUtils.isEmpty(topic)){
result = "agrs is Empty";
return result;
}
//其他参数
String source = tools.get(KEY_SOURCE);
if(!SOURCE_PRODUCT.equalsIgnoreCase(source) && !SOURCE_PAY.equalsIgnoreCase(source)){
result = "source is error .source must be ['product' or 'pay']";
return result;
}
Integer count = tools.getInt(KEY_COUNT);
if (null == count){
result = "count is empty";
return result;
}else{
if (count > COUNT_MAX || count < COUNT_MIN){
result = "count is unbound[" + COUNT_MIN + "~"+ COUNT_MAX +"]";
return result;
}
}
Integer sleep = tools.getInt(KEY_SLEEP);
if( null == sleep){
result = " sleep is empty";
return result;
} else{
if (sleep > SLEEP_MAX || sleep < SLEEP_MIN){
result = "sleep is unbound[" + SLEEP_MIN + "~"+ SLEEP_MAX +"]";
return result;
}
}
return result;
}
//选择造数
public static void chooseFun(String[] args){
//通过加载args参数列表获取参数工具类对象
QParameterTool tools = QParameterTool.fromArgs(args);
String topic = tools.get(KEY_TOPIC);
String source = tools.get(KEY_SOURCE);
Integer count = tools.getInt(KEY_COUNT);
Integer sleep = tools.getInt(KEY_SLEEP);
if (SOURCE_PRODUCT.equalsIgnoreCase(source)){
//创建关于产品的数据
makeTravelProductData(topic,count,sleep);
}else if(SOURCE_PAY.equalsIgnoreCase(source)){
}
}
//创造产品订单数据
public static void makeTravelProductData(String topic, Integer count, Integer sleep) {
//发送序列化对象
String dateFormatter = CommonConstant.FORMATTER_YYYYMMDDHHMMDD;
String dayFormatter = CommonConstant.FORMATTER_YYYYMMDD;
ChronoUnit chronoUnit = ChronoUnit.MINUTES;
ChronoUnit daysChronoUnit = ChronoUnit.DAYS;
//辅助数据
//2.1地区信息
Listregions = getRegions();
//2.2 酒店信息
Map pubs = getPubMappingPro();
//2.3产品信息
List productIDs = getProducts();
}
//旅游产品信息
private static List getProducts() {
if (CollectionUtils.isEmpty(products)){
try{
List> productDatas = CSVUtil.readCSVFile(CSVUtil.PRODUCT_FILE,CSVUtil.QUOTE_COMMON);
if(CollectionUtils.isNotEmpty(productDatas)){
for (Map product:productDatas){
String productID = product.getOrDefault(PRODUCT_KEY_ID,"");
products.add(productID);
}
}
}catch (Exception e){
log.error("TravelOrderHelper 's products errors :" + e);
}
}
return products;
}
//旅游产品与酒店映射关系
private static Map getPubMappingPro() {
//表示没有初始化
if(MapUtils.isEmpty(proMappingPub)){
try{
List> pubDatas = CSVUtil.readCSVFile(CSVUtil.PUB_FILE,CSVUtil.QUOTE_COMMON);
//校验
if(CollectionUtils.isNotEmpty(pubDatas)){
//遍历
for (Map pub : pubDatas){
String pubId = pub.getOrDefault(PUB_KEY_ID,"");
String[] pps = pubId.split("\\|");
proMappingPub.put(pps[0],pubId);
}
}
}catch (Exception e){
log.error("TravelOrderelper 's pub errors:" + e);
}
}
return proMappingPub;
}
//获取到地区列表
private static List getRegions() {
//如果集合为空，就是第一次获取地区，需要初始化地区信息
if (CollectionUtils.isEmpty(regions)){
try{
//就是一个路径，一个是分隔符
List > regionsDatas = CSVUtil.readCSVFile(CSVUtil.REGION_FILE,CSVUtil.QUOTE_COMMON);
//将regionDatas中Map封装到regionDo对象
if(CollectionUtils.isNotEmpty(regionsDatas)){
for (Map region : regionsDatas){
RegionDo regionDo = new RegionDo();
regionDo.setRegionCity(region.getOrDefault(REGION_KEY_CITY, ""));
regionDo.setRegionCityDesc(region.getOrDefault(REGION_KEY_CITY_DESC, ""));
regionDo.setRegionCode(region.getOrDefault(REGION_KEY_CODE, ""));
regionDo.setRegionCodeDesc(region.getOrDefault(REGION_KEY_CODE_DESC, ""));
regionDo.setRegionProvince(region.getOrDefault(REGION_KEY_PROVINCE, ""));
regionDo.setRegionProvinceDesc(region.getOrDefault(REGION_KEY_PROVINCE_DESC, ""));
regions.add(regionDo);
}
}
}catch(Exception e){
log.error("TravelOrderHelper 's region errors : " +e);
}
}
return regions;
}
}
微信扫一扫
支付宝扫一扫
评论列表（0条）