package com.formulasearchengine.mathosphere.mlp;

import com.formulasearchengine.mathosphere.mlp.cli.TagsCommandConfig;
import com.formulasearchengine.mathosphere.mlp.contracts.TagExtractionMapper;
import com.formulasearchengine.mathosphere.mlp.contracts.TextExtractorMapper;
import com.formulasearchengine.mathosphere.mlp.pojos.MathTag;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;

public class WikiTagExtractor {


    public static void run(TagsCommandConfig config) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        final DataSource<String> dump = FlinkMlpRelationFinder.readWikiDump(config, env);
        dump
                .flatMap(new TextExtractorMapper())
                .flatMap(new TagExtractionMapper(config))
                .distinct(MathTag::getContentHash)
                .map(MathTag::toJson)
                .writeAsText(config.getOutputDir() + "/formulae.txt");
        env.execute();
    }
}

@@ -721,16 +721,73 @@
 , env);%0A
+        //DataSet%3CString%3E text = env.fromElements(%22%5B %22);%0A
         
@@ -1014,16 +1014,25 @@
 .writeAs
+Formatted
 Text(con
@@ -1067,16 +1067,65 @@
 lae.txt%22
+, s -%3E s + %22,%22)%0A                .setParallelism(1
 );%0A     

Add commas to JSON object list