文章目录一、需求描述二、实现1. 主要连接器依赖2. udf正则获取字段数据3. sql一、需求描述通过正则获取HDFS log日志中指定字段的数据并输出到HIVE表中。flink版本1.12.7。二、实现1. 主要连接器依赖!-- filesystem连接器 --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-connector-files/artifactIdversion${flink.version}/version/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-csv/artifactIdversion${flink.version}/version/dependency2. udf正则获取字段数据packagecom.gao.function;importorg.apache.flink.table.annotation.DataTypeHint;importorg.apache.flink.table.annotation.FunctionHint;importorg.apache.flink.table.functions.TableFunction;importorg.apache.flink.types.Row;importorg.apache.flink.types.RowKind;importjava.util.ArrayList;importjava.util.List;importjava.util.regex.Matcher;importjava.util.regex.Pattern;publicclassRegexpExtractTupleUdfextendsTableFunctionRow{FunctionHint(outputDataTypeHint(ROWarr ARRAYSTRING))publicvoideval(Stringraw,Stringpattern){if(rawnull||patternnull){return;}ListStringlistnewArrayList();PatternrPattern.compile(pattern);Matchermr.matcher(raw);if(m.find()){for(inti1;im.groupCount();i){list.add(m.group(i));}}String[]arrlist.toArray(newString[0]);RowKindrowKindRowKind.fromByteValue((byte)0);RowrownewRow(rowKind,1);row.setField(0,arr);collect(row);}}3. sqlCREATETEMPORARYSYSTEMFUNCTIONregexp_extract_tupleAScom.gao.function.RegexpExtractTupleUdf;CREATETABLEsource(raw_messagestring)WITH(connectorfilesystem,pathhdfs:xxx,formatraw);CREATETABLEsink(gifstring,refererstring,midstring)WITH(connectorfilesystem,pathhdfs://xxx,formatcsv,csv.field-delimiter\t,--字段分隔符用于hive字段识别使用\tcsv.disable-quote-charactertrue,--字符串类型的数据不加双引号输出csv.null-literalnull-- 数据为空是 输出为null);CREATETEMPORARYVIEWsource_1ASselectraw_message,REGEXP_EXTRACT(raw_message,(\?||)gif([^ \]*),2)asgif-- 通过正则取gif字段内容,REGEXP_EXTRACT(raw_message,(\?||)_referer([^ \]*),2)asreferer,REGEXP_EXTRACT(raw_message,(\?||)mid([^ \]*),2)asmidfromsource S;insertintosinkselectgifasgif,refererasreferer,midasmidfromsource_1;
