parsing un structured logs in to dataset in flink

user_3607698 0
Viewed 76

I have some logs like below

2021-08-16 16:56:37 info [614289000000000020ab22baa4a1cfb7,4c8b36453209aef4] -- [main] c.g.c.uservalidatorfactoryimpl - {"userid": "1", "msg": "action", "status": "rcvd"}
2021-08-16 16:56:38 info [614289000000000020ab22baa4a1cfb7,4c8b36453209aef4] -- [main] c.g.c.uservalidatorfactoryimpl - {"userid": "1", "msg": "response", "status": "processed"}

I need to read this from the file and convert to dataset which looks something like below

2021-08-16 16:56:37,[614289000000000020ab22baa4a1cfb7,4c8b36453209aef4],  1 , action , rcvd  
2021-08-16 16:56:38,[614289000000000020ab22baa4a1cfb7,4c8b36453209aef4],  1 , response , processed   


 val dataset = env.readTextFile("text.log")
    dataset.print()

Since the log file is not structured, is it possible to parse and clean the logs into CSV format using any of the flink APIS?

it will be really helpful if someone can through light on any other best way convert in to csv format so that it can be loaded for further processing?

apache-flink
1 Answers

Flink has integrated RowCsvInputFormat to read csv format files, and supports setting the file path and csv schema information

Path PATH = new Path("an/ignored/file/");
String fileContent = "abc|def|ghijk\nabc||hhg\n|||\n||";

TypeInformation[] fieldTypes =
        new TypeInformation[] {
            BasicTypeInfo.STRING_TYPE_INFO,
            BasicTypeInfo.STRING_TYPE_INFO,
            BasicTypeInfo.STRING_TYPE_INFO
        };

RowCsvInputFormat.Builder builder =
        RowCsvInputFormat.builder(new RowTypeInfo(fieldTypes), PATH).setFieldDelimiter('|');
RowCsvInputFormat format = builder.build();

env.createInput(format).print()
user_168675870
Related