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 08:36:43 UTC
(stormcrawler) branch FileSpoutCompressedFiles created (now 21057246)
This is an automated email from the ASF dual-hosted git repository.
jnioche pushed a change to branch FileSpoutCompressedFiles
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
at 21057246 Add support for gzip and bz2 files + fix javadoc and prevent possible leak
This branch includes the following new commits:
new 21057246 Add support for gzip and bz2 files + fix javadoc and prevent possible leak
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(stormcrawler) 01/01: Add support for gzip and bz2 files + fix javadoc and prevent possible leak
Posted by jn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jnioche pushed a commit to branch FileSpoutCompressedFiles
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
commit 2105724668f2c06f03ab157c207b60086d1c61be
Author: Julien Nioche <ju...@digitalpebble.com>
AuthorDate: Mon Jun 15 18:23:37 2026 +0100
Add support for gzip and bz2 files + fix javadoc and prevent possible leak
Signed-off-by: Julien Nioche <ju...@digitalpebble.com>
---
core/pom.xml | 5 ++
.../org/apache/stormcrawler/spout/FileSpout.java | 60 +++++++++++++---------
.../apache/stormcrawler/spout/FileSpoutTest.java | 58 +++++++++++++++++++++
3 files changed, 98 insertions(+), 25 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..cdf22d8e 100644
--- a/core/src/main/java/org/apache/stormcrawler/spout/FileSpout.java
+++ b/core/src/main/java/org/apache/stormcrawler/spout/FileSpout.java
@@ -17,21 +17,16 @@
package org.apache.stormcrawler.spout;
-import java.io.BufferedReader;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStreamReader;
+import java.io.*;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
-import java.util.Collections;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Queue;
+import java.util.*;
+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 +41,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 +130,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 FileInputStream(inputPath.toFile());
+ try {
+ String fileLower = file.toLowerCase();
+ if (fileLower.endsWith(".gz") || fileLower.endsWith(".gzip")) {
+ is = new GZIPInputStream(is);
+ } else if (fileLower.endsWith(".bz2")) {
+ is = new BZip2CompressorInputStream(is);
+ }
+ currentBuffer =
+ new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8));
+ } catch (IOException e) {
+ is.close();
+ throw e;
+ }
}
String line = null;
@@ -228,7 +229,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(