commenct

laglangyue · Aug 17, 2023 · 2378e77 · 2378e77
1 parent 05ed58e
commit 2378e77
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 1 deletion.
diff --git a/...4/src/main/java/org/apache/seatunnel/translation/spark/source/SeaTunnelSourceSupport.java b/...4/src/main/java/org/apache/seatunnel/translation/spark/source/SeaTunnelSourceSupport.java
@@ -86,7 +86,7 @@ public MicroBatchReader createMicroBatchReader(
         // 创建微批 读取器，spark streaming，即将被删除
         SeaTunnelSource<SeaTunnelRow, ?, ?> seaTunnelSource = getSeaTunnelSource(options);
         Integer parallelism = options.getInt(CommonOptions.PARALLELISM.key(), 1);
-        // checkpoint
+        // checkpoint 配置
         Integer checkpointInterval =
                 options.getInt(
                         EnvCommonOptions.CHECKPOINT_INTERVAL.key(), CHECKPOINT_INTERVAL_DEFAULT);

diff --git a/...va/org/apache/seatunnel/translation/spark/source/partition/micro/MicroBatchPartition.java b/...va/org/apache/seatunnel/translation/spark/source/partition/micro/MicroBatchPartition.java
@@ -29,6 +29,9 @@
 import org.apache.spark.sql.sources.v2.reader.InputPartition;
 import org.apache.spark.sql.sources.v2.reader.InputPartitionReader;
 
+/**
+ * 一个分区读取器
+ */
 public class MicroBatchPartition implements InputPartition<InternalRow> {
     protected final SeaTunnelSource<SeaTunnelRow, ?, ?> source;
     protected final Integer parallelism;

diff --git a/...he/seatunnel/translation/spark/source/reader/micro/ParallelMicroBatchPartitionReader.java b/...he/seatunnel/translation/spark/source/reader/micro/ParallelMicroBatchPartitionReader.java
@@ -42,6 +42,9 @@
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
+/**
+ * 这个类只处理任务状态,真正的读取在父类
+ */
 public class ParallelMicroBatchPartitionReader extends ParallelBatchPartitionReader {
     protected static final Integer CHECKPOINT_SLEEP_INTERVAL = 10;
     protected static final Integer CHECKPOINT_RETRIES = 3;
@@ -91,6 +94,7 @@ protected void prepare() {
 
     protected FileSystem getFileSystem()
             throws URISyntaxException, IOException, InterruptedException {
+        // 状态读取
         Configuration configuration = new Configuration();
         configuration.set("fs.defaultFS", hdfsRoot);
         if (StringUtils.isNotBlank(hdfsUser)) {