THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
Current state: Under Discussion
Discussion thread: discuss mail
JIRA:
Jira | ||||||
---|---|---|---|---|---|---|
|
PULL REQUEST: https://github.com/apache/kafka/pull/12219 ( before review PR https://github.com/apache/kafka/pull/7965 )
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
- Parsing AccessLog
Input
Code Block |
---|
{ "message" : "111.61.73.113 - - [08/Aug/2019:18:15:29 +0900] \"OPTIONS /api/v1/service_config HTTP/1.1\" 200 - 101989 \"http://local.test.com/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36\"" } |
Output
Code Block |
---|
{ "IP" : "111.61.73.113" ,"RemoteUser" : "-" ,"AuthedRemoteUser" : "-" ,"DateTime" : "08/Aug/2019:18:15:29 +0900" ,"Method" : "OPTIONS" ,"Request" : "/api/v1/service_config" ,"Protocol" : "HTTP/1.1" ,"Response" : "200" ,"BytesSent" : "-" ,"Ms" : "101989" ,"Referrer" : "http://local.test.com" ,"UserAgent" : "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_14_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/75.0.3770.142 Safari/537.36"} |
...
2. Parsing URL
Input
Code Block |
---|
{ "url" : "https://kafka.apache.org/documentation/#connect" } |
Output
Code Block |
---|
{ "protocol" : "https" ,"domain" : "kafka.apache.org" ,"path" : "documentation/#connect" } |
...
Code Block |
---|
"transforms": "RegexTransform",
"transforms.RegexTransform.type": "org.apache.kafka.connect.transforms.ToStructByRegexTransform$Value",
"transforms.RegexTransform.regex": "^([\\d.]+) (\\S+) (\\S+) \\[([\\w:/]+\\s[+\\-]\\d{4})\\] \"(GET|POST|OPTIONS|HEAD|PUT|DELETE|PATCH) (.+?) (.+?)\" (\\d{3}) ([0-9|-]+) ([0-9|-]+) \"([^\"]+)\" \"([^\"]+)\""
"transforms.RegexTransform.mapping": "IP,RemoteUser,AuthedRemoteUser,DateTime,Method,Request,Protocol,Response,BytesSent,Ms:NUMBER,Referrer,UserAgent" |
Usecase2. Transform Config
...
Code Block | ||
---|---|---|
| ||
public abstract class ToStructByRegexTransfo
rm<R extends ConnectRecord<R>> implements Transformation<R> {
public static final String OVERVIEW_DOC = "Generate key/value Struct objects supported by ordered Regex Group"
+ "<p/>Use the concrete transformation type designed for the record key (<code>" + Key.class.getName() + "</code>) "
+ "or value (<code>" + Value.class.getName() + "</code>).";
private static final String TYPE_DELIMITER = ":";
private interface ConfigName {
String REGEX = "regex";
String MAPPING_KEY = "mapping";
}
public static final ConfigDef CONFIG_DEF = new ConfigDef()
.define(ConfigName.REGEX, ConfigDef.Type.STRING, ConfigDef.NO_DEFAULT_VALUE, new GroupRegexValidator(), ConfigDef.Importance.MEDIUM,
"String Regex Group Pattern.")
.define(ConfigName.MAPPING_KEY, ConfigDef.Type.LIST, ConfigDef.NO_DEFAULT_VALUE, ConfigDef.Importance.MEDIUM,
"Ordered Regex Group Mapping Keys ( with :{TYPE} )");
private static final String PURPOSE = "Transform Struct by regex group mapping"
... |
...
2. main functions interface
Code Block | ||
---|---|---|
| ||
@Override public R apply(R record) { if (operatingSchema(record) == null) { return applySchemaless(record); } else { return applyWithSchema(record); } } private R applySchemaless(R record) { ... } private R applyWithSchema(R record) { ... } |
...