You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@wayang.apache.org by rp...@apache.org on 2022/12/09 14:47:03 UTC
[incubator-wayang] 04/04: Pipeline running locally
This is an automated email from the ASF dual-hosted git repository.
rpardomeza pushed a commit to branch WAYANG-agoraeo
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit 4bf96a1ad405baa747a14f86a333a32e480a51ed
Author: rodrigopardomeza <ro...@gmail.com>
AuthorDate: Fri Dec 9 15:46:45 2022 +0100
Pipeline running locally
---
.../main/java/org/apache/wayang/agoraeo/Main.java | 122 +++++++++++++++++----
1 file changed, 103 insertions(+), 19 deletions(-)
diff --git a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java
index 02276f77..6729d56d 100644
--- a/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java
+++ b/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/Main.java
@@ -9,6 +9,7 @@ import java.util.stream.Stream;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.wayang.agoraeo.iterators.FileIteratorSentinelDownload;
import org.apache.wayang.agoraeo.iterators.IteratorSentinelDownload;
+import org.apache.wayang.agoraeo.iterators.StringIteratorSentinelDownload;
import org.apache.wayang.agoraeo.operators.basic.SentinelSource;
import org.apache.wayang.basic.data.Tuple2;
import org.apache.wayang.basic.operators.*;
@@ -32,16 +33,25 @@ import org.apache.wayang.spark.Spark;
public class Main {
public static void main(String[] args) {
+ String downloader_location = "/Users/rodrigopardomeza/PycharmProjects/MinimalDownload/";
+ String images_folder = "images/";
+ String download_program = "single_mirror_manager.py";
+
String input = "file:///Users/rodrigopardomeza/files/file.txt";
- String output = "file:///Users/rodrigopardomeza/files/output-java.txt";
+ String output = "file:///Users/rodrigopardomeza/files/sen2cor-output-agoraeo.txt";
System.out.println("Hello AgoraEO!");
- String cmd = "echo \"file:///Users/rodrigopardomeza/files/file.txt\"";
+// String cmd = "echo \"file:///Users/rodrigopardomeza/files/file.txt\"";
+// String cmd = "echo \"~/tu-berlin/agoraeo/images/33UVT/S2B_MSIL1C_20221119T101229_N0400_R022_T33UVT_20221119T104925.SAFE\"";
+// String cmd = "/Users/rodrigopardomeza/tu-berlin/agoraeo/agoraeo/wayang-plugins/wayang-agoraeo/src/main/java/org/apache/wayang/agoraeo/workers/name.sh";
+ String cmd =
+ "/Users/rodrigopardomeza/PycharmProjects/MinimalDownload/venv/bin/python /Users/rodrigopardomeza/PycharmProjects/MinimalDownload/single_mirror_manager.py --user rpardomeza --password 12c124ccb2 --url https://scihub.copernicus.eu/dhus --from NOW-30DAY --to NOW --order 33UVT";
- WayangPlan w = createWayangPlan(cmd, output);
+ WayangPlan w = alternative2WayangPlan(cmd, output);
WayangContext wayangContext = new WayangContext();
wayangContext.register(Java.basicPlugin());
+ wayangContext.register(WayangAgoraEO.javaPlugin());
// wayangContext.register(Spark.basicPlugin());
// for (String platform : args[0].split(",")) {
// switch (platform) {
@@ -62,6 +72,91 @@ public class Main {
}
+ public static WayangPlan alternativeWayangPlan(
+ String cmd,
+ String outputFileUrl
+ ) {
+ IteratorSentinelDownload<File> iter = new FileIteratorSentinelDownload("Sentinel 2 - API", cmd);
+
+ SentinelSource<File> source = new SentinelSource<>(iter, File.class);
+
+ /*TODO Replace String to call for Sen2Cor and retrieve output of the command*/
+ MapOperator<File, String> files_names = new MapOperator<>(
+ t -> t.getAbsolutePath(),
+ File.class,
+ String.class
+ );
+
+// FlatMapOperator<File, String> files_names = new FlatMapOperator<>(
+// t -> {
+// try {
+// String command = "~/Downloads/Sen2Cor-02.10.01-Darwin64/bin/L2A_Process\n" +
+// "~/tu-berlin/agoraeo/images/33UVT/S2B_MSIL1C_20221119T101229_N0400_R022_T33UVT_20221119T104925.SAFE\n" +
+// "--output_dir ~/tu-berlin/agoraeo/images/33UVT/L2A/";
+// Process process = Runtime.getRuntime().exec(command);
+// Iterator<String> input = new BufferedReader(
+// new InputStreamReader(
+// process.getInputStream()
+// )
+// ).lines().iterator();
+// return () -> input;
+// } catch (IOException e) {
+// throw new RuntimeException(e);
+// }
+// },
+// File.class,
+// String.class
+// );
+
+
+ TextFileSink<String> sink = new TextFileSink<>(outputFileUrl, String.class);
+
+ source.connectTo(0,files_names,0);
+ files_names.connectTo(0,sink,0);
+
+
+ return new WayangPlan(sink);
+ }
+
+ public static WayangPlan alternative2WayangPlan(
+ String cmd,
+ String outputFileUrl
+ ) {
+ IteratorSentinelDownload<String> iter = new StringIteratorSentinelDownload("Sentinel 2 - API", cmd);
+
+ SentinelSource<String> source = new SentinelSource<>(iter, String.class);
+
+ FlatMapOperator<String, String> files_names = new FlatMapOperator<>(
+ t -> {
+ try {
+ String command = "/Users/rodrigopardomeza/Downloads/Sen2Cor-02.10.01-Darwin64/bin/L2A_Process\n" +
+ t + "\n" +
+ "--output_dir /Users/rodrigopardomeza/tu-berlin/agoraeo/images/pipeline-result/";
+ Process process = Runtime.getRuntime().exec(command);
+ Iterator<String> input = new BufferedReader(
+ new InputStreamReader(
+ process.getInputStream()
+ )
+ ).lines().iterator();
+ return () -> input;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ },
+ String.class,
+ String.class
+ );
+
+
+ TextFileSink<String> sink = new TextFileSink<>(outputFileUrl, String.class);
+
+ source.connectTo(0,files_names,0);
+ files_names.connectTo(0,sink,0);
+
+
+ return new WayangPlan(sink);
+ }
+
public static WayangPlan basicWayangPlan(
String inputFileUrl,
String outputFileUrl
@@ -97,18 +192,14 @@ public class Main {
public static WayangPlan createWayangPlan(
String cmd,
String outputFileUrl
-// ,
-// Collection<Tuple2<String, Integer>> collector
) {
- Collection<Tuple2<String, Integer>> collector = new LinkedList<>();
-
IteratorSentinelDownload<File> iter = new FileIteratorSentinelDownload("Sentinel 2 - API", cmd);
/* Might replay the name of the downloaded file */
SentinelSource<File> source = new SentinelSource<>(iter, File.class);
- FlatMapOperator<File, String> files_lines = new FlatMapOperator<File, String>(
+ FlatMapOperator<File, String> files_lines = new FlatMapOperator<>(
t -> {
try {
final InputStream inputStream = Files.newInputStream(t.toPath());
@@ -121,17 +212,14 @@ public class Main {
File.class,
String.class
);
+ files_lines.setName("files giving lines");
- FlatMapOperator<String, String> words = new FlatMapOperator<String, String>(
+ FlatMapOperator<String, String> words = new FlatMapOperator<>(
line -> Arrays.asList(line.split("\\W+")),
String.class,
String.class
);
-
- files_lines.setName("files giving lines");
-
-
-
+ words.setName("words");
// for each word transform it to lowercase and output a key-value pair (word, 1)
MapOperator<String, Tuple2<String, Integer>> mapOperator = new MapOperator<>(
@@ -158,11 +246,7 @@ public class Main {
);
reduceByOperator.setName("Add counters");
-
- LocalCallbackSink<Tuple2<String, Integer>> sink = LocalCallbackSink.createCollectingSink(
- collector,
- DataSetType.createDefaultUnchecked(Tuple2.class)
- );
+ TextFileSink<Tuple2> sink = new TextFileSink<>(outputFileUrl, Tuple2.class);
sink.setName("Collect result");