You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stormcrawler.apache.org by jn...@apache.org on 2026/06/16 11:16:20 UTC
(stormcrawler) branch main updated: FileSpout: add support for gzip and bz2 files (#1952)
This is an automated email from the ASF dual-hosted git repository.
jnioche pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
The following commit(s) were added to refs/heads/main by this push:
new 62177234 FileSpout: add support for gzip and bz2 files (#1952)
62177234 is described below
commit 62177234de21064b42735d01249b8ce657c092a8
Author: Julien Nioche <ju...@digitalpebble.com>
AuthorDate: Tue Jun 16 12:16:17 2026 +0100
FileSpout: add support for gzip and bz2 files (#1952)
* Add support for gzip and bz2 files + fix javadoc and prevent possible leak
Signed-off-by: Julien Nioche <ju...@digitalpebble.com>
* comply to checkstyle
Signed-off-by: Julien Nioche <ju...@digitalpebble.com>
* buffered input + handle concats for bz2
Signed-off-by: Julien Nioche <ju...@digitalpebble.com>
---------
Signed-off-by: Julien Nioche <ju...@digitalpebble.com>
---
core/pom.xml | 5 ++
.../org/apache/stormcrawler/spout/FileSpout.java | 52 +++++++++++++------
.../apache/stormcrawler/spout/FileSpoutTest.java | 58 ++++++++++++++++++++++
3 files changed, 99 insertions(+), 16 deletions(-)
diff --git a/core/pom.xml b/core/pom.xml
index 12ad9985..6d12c427 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -127,6 +127,11 @@ under the License.
<version>${tika.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-compress</artifactId>
+ </dependency>
+
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
diff --git a/core/src/main/java/org/apache/stormcrawler/spout/FileSpout.java b/core/src/main/java/org/apache/stormcrawler/spout/FileSpout.java
index 6615ea51..8e2a9edd 100644
--- a/core/src/main/java/org/apache/stormcrawler/spout/FileSpout.java
+++ b/core/src/main/java/org/apache/stormcrawler/spout/FileSpout.java
@@ -17,9 +17,11 @@
package org.apache.stormcrawler.spout;
+import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
+import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
@@ -30,8 +32,11 @@ import java.nio.file.Paths;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Queue;
+import java.util.zip.GZIPInputStream;
+import org.apache.commons.compress.compressors.bzip2.BZip2CompressorInputStream;
import org.apache.commons.lang3.StringUtils;
import org.apache.storm.spout.Scheme;
import org.apache.storm.spout.SpoutOutputCollector;
@@ -46,27 +51,24 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Reads the lines from a UTF-8 file and use them as a spout. Load the entire content into memory.
- * Uses StringTabScheme to parse the lines into URLs and Metadata, generates tuples on the default
- * stream unless withDiscoveredStatus is set to true.
+ * Reads the lines from a UTF-8 file and use them as a spout. The spout reads files in chunks of
+ * 10,000 lines, keeping memory usage very low even for extremely large files with millions of seed
+ * URLs. Uses StringTabScheme to parse the lines into URLs and Metadata, generates tuples on the
+ * default stream unless withDiscoveredStatus is set to true.
*/
public class FileSpout extends BaseRichSpout {
public static final int BATCH_SIZE = 10000;
public static final Logger LOG = LoggerFactory.getLogger(FileSpout.class);
-
- protected SpoutOutputCollector collector;
-
private final Queue<String> inputFiles;
- private BufferedReader currentBuffer;
-
+ protected SpoutOutputCollector collector;
protected Scheme scheme = new StringTabScheme();
-
protected LinkedList<byte[]> buffer = new LinkedList<>();
protected boolean active;
- private boolean withDiscoveredStatus = false;
protected int totalTasks;
protected int taskIndex;
+ private BufferedReader currentBuffer;
+ private boolean withDiscoveredStatus = false;
/**
* @param dir containing the seed files
@@ -138,11 +140,20 @@ public class FileSpout extends BaseRichSpout {
return;
}
Path inputPath = Paths.get(file);
- currentBuffer =
- new BufferedReader(
- new InputStreamReader(
- new FileInputStream(inputPath.toFile()),
- StandardCharsets.UTF_8));
+ InputStream is = new BufferedInputStream(new FileInputStream(inputPath.toFile()));
+ try {
+ String fileLower = file.toLowerCase(Locale.ROOT);
+ if (fileLower.endsWith(".gz") || fileLower.endsWith(".gzip")) {
+ is = new GZIPInputStream(is);
+ } else if (fileLower.endsWith(".bz2")) {
+ is = new BZip2CompressorInputStream(is, true);
+ }
+ currentBuffer =
+ new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8));
+ } catch (IOException e) {
+ is.close();
+ throw e;
+ }
}
String line = null;
@@ -228,7 +239,16 @@ public class FileSpout extends BaseRichSpout {
}
@Override
- public void close() {}
+ public void close() {
+ if (currentBuffer != null) {
+ try {
+ currentBuffer.close();
+ } catch (IOException e) {
+ LOG.error("Exception thrown when closing current buffer", e);
+ }
+ currentBuffer = null;
+ }
+ }
@Override
public void activate() {
diff --git a/core/src/test/java/org/apache/stormcrawler/spout/FileSpoutTest.java b/core/src/test/java/org/apache/stormcrawler/spout/FileSpoutTest.java
index 8d6b03db..e0447b28 100644
--- a/core/src/test/java/org/apache/stormcrawler/spout/FileSpoutTest.java
+++ b/core/src/test/java/org/apache/stormcrawler/spout/FileSpoutTest.java
@@ -22,6 +22,8 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.util.List;
import java.util.Map;
@@ -101,6 +103,62 @@ class FileSpoutTest {
assertEquals(Status.DISCOVERED, tuple.get(2));
}
+ @Test
+ void testGzipSeedFile(@org.junit.jupiter.api.io.TempDir Path tempDir) throws Exception {
+ Path gzipFile = tempDir.resolve("seed-list.txt.gz");
+ try (java.io.OutputStream os = Files.newOutputStream(gzipFile);
+ java.util.zip.GZIPOutputStream gzos = new java.util.zip.GZIPOutputStream(os)) {
+ gzos.write("https://stormcrawler.apache.org\n".getBytes(StandardCharsets.UTF_8));
+ gzos.write("https://github.com\n".getBytes(StandardCharsets.UTF_8));
+ }
+
+ final FileSpout spout = new FileSpout(gzipFile.toAbsolutePath().toString());
+ final FileSpoutOutputCollectorMock collectorMock = new FileSpoutOutputCollectorMock();
+ spout.open(Map.of(), new FileSpoutTopologyContextMock(), collectorMock);
+ spout.activate();
+
+ // 1st tuple
+ spout.nextTuple();
+ List<Object> tuple = collectorMock.getTuple();
+ assertNotNull(tuple);
+ assertEquals("https://stormcrawler.apache.org", tuple.get(0));
+
+ // 2nd tuple
+ spout.nextTuple();
+ tuple = collectorMock.getTuple();
+ assertNotNull(tuple);
+ assertEquals("https://github.com", tuple.get(0));
+ }
+
+ @Test
+ void testBzip2SeedFile(@org.junit.jupiter.api.io.TempDir Path tempDir) throws Exception {
+ Path bzip2File = tempDir.resolve("seed-list.txt.bz2");
+ try (java.io.OutputStream os = Files.newOutputStream(bzip2File);
+ org.apache.commons.compress.compressors.bzip2.BZip2CompressorOutputStream bzos =
+ new org.apache.commons.compress.compressors.bzip2
+ .BZip2CompressorOutputStream(os)) {
+ bzos.write("https://stormcrawler.apache.org\n".getBytes(StandardCharsets.UTF_8));
+ bzos.write("https://github.com\n".getBytes(StandardCharsets.UTF_8));
+ }
+
+ final FileSpout spout = new FileSpout(bzip2File.toAbsolutePath().toString());
+ final FileSpoutOutputCollectorMock collectorMock = new FileSpoutOutputCollectorMock();
+ spout.open(Map.of(), new FileSpoutTopologyContextMock(), collectorMock);
+ spout.activate();
+
+ // 1st tuple
+ spout.nextTuple();
+ List<Object> tuple = collectorMock.getTuple();
+ assertNotNull(tuple);
+ assertEquals("https://stormcrawler.apache.org", tuple.get(0));
+
+ // 2nd tuple
+ spout.nextTuple();
+ tuple = collectorMock.getTuple();
+ assertNotNull(tuple);
+ assertEquals("https://github.com", tuple.get(0));
+ }
+
private Path getPath(String resource) throws URISyntaxException {
return Path.of(
Objects.requireNonNull(