Repository: scrapy/scrapy
Pull request: Feed exports: add batch deliveries
Fixes #4250

Problem statement:
Batch deliveries for long running crawlers
## Summary

Add a new setting `FEED_STORAGE_BATCH` that will deliver a file whenever `item_scraped_count` reaches a multiple of that number.

## Motivation

For long running jobs (say we are consuming inputs from a working queue) we may want partial results instead of waiting for a long batch to finish.

## Describe alternatives you've considered

Of course we can stop and restart a spider every now and then.
However, a simpler approach is to have it running as long as required, but delivering partial results.

Hints from issue discussion:
This will need unique file names, right?
Indeed.
When I wrote it at first, I was thinking about templates.
One simple alternative, would be to force `%(time)s` when this is set, so we are sure the file names will be different
Anyone working on this? Wanted to get started contributing to this project by jumping into this issue if no one else is.
@alake16
Hi, I have already started working on this issue, but you can also start if you want.
Once you have something, you can create a draft PR in Github so we can follow through there.
Also consider an alternative to deliver data at every N items or X minutes.
@BroodingKangaroo how's it going for you? Blocked on anything at the moment?
@alake16 I am in the process of understanding how scrapy works inside and how to do what is required. :smiley_cat: 
@ejulio We should merge and remove partial results after finishing work, should we?
@BroodingKangaroo , my idea is that it should be partial deliveries.
No file deletion/aggregation, as that would be basically the current behavior.
The idea here is to be able to deliver files before a run finishes, so we don't need to wait days/hours for a result.
Hi @ejulio ! =)
Could you review my PR?

Also I want to introduce myself.
My name is Maxim Halavaty and I am a third year student of Belarusian State University at the faculty of Applied mathematics and computer science. I would be happy to help develop your project during the GSOC 2020.

I have some questions regarding this issue. 
1. Should we create new slot for each new file/partial?
2. How should we name partial files. My spider (with realization that I have in PR) for this input
`scrapy crawl quotes --nolog -o res/test.json` with FEED_STORAGE_BATCH = 1000
creates such file tree:
res-->
-------->test.2020-03-18T14:43:58.896341.json
-------->test.2020-03-18T14:44:01.465171.json
-------->test.2020-03-18T14:44:02.326922.json
each file has 1000 entries, and last one has 395 (<=1000).
@BroodingKangaroo , great!
For GSoC, this is the first contribution which we ask, but you should write a project proposal so we can evaluate it.
You can write a draft and we can help you build it for final submission.

Commit message: Add quotes to example in docs

Patch to review:
diff --git a/docs/topics/feed-exports.rst b/docs/topics/feed-exports.rst
index 24d69040c32..7e91b365dfc 100644
--- a/docs/topics/feed-exports.rst
+++ b/docs/topics/feed-exports.rst
@@ -220,6 +220,7 @@ These are the settings used for configuring the feed exports:
  * :setting:`FEED_STORAGE_FTP_ACTIVE`
  * :setting:`FEED_STORAGE_S3_ACL`
  * :setting:`FEED_EXPORTERS`
+ * :setting:`FEED_EXPORT_BATCH_ITEM_COUNT`
 
 .. currentmodule:: scrapy.extensions.feedexport
 
@@ -271,6 +272,7 @@ as a fallback value if that key is not provided for a specific feed definition.
 * ``fields``: falls back to :setting:`FEED_EXPORT_FIELDS`
 * ``indent``: falls back to :setting:`FEED_EXPORT_INDENT`
 * ``store_empty``: falls back to :setting:`FEED_STORE_EMPTY`
+* ``batch_item_count``: falls back to :setting:`FEED_EXPORT_BATCH_ITEM_COUNT`
 
 .. setting:: FEED_EXPORT_ENCODING
 
@@ -429,3 +431,45 @@ format in :setting:`FEED_EXPORTERS`. E.g., to disable the built-in CSV exporter
 .. _Amazon S3: https://aws.amazon.com/s3/
 .. _botocore: https://github.com/boto/botocore
 .. _Canned ACL: https://docs.aws.amazon.com/AmazonS3/latest/dev/acl-overview.html#canned-acl
+
+.. setting:: FEED_EXPORT_BATCH_ITEM_COUNT
+
+FEED_EXPORT_BATCH_ITEM_COUNT
+-----------------------------
+Default: ``0``
+
+If assigned an integer number higher than ``0``, Scrapy generates multiple output files
+storing up to the specified number of items in each output file.
+
+When generating multiple output files, you must use at least one of the following
+placeholders in the feed URI to indicate how the different output file names are
+generated:
+
+* ``%(batch_time)s`` - gets replaced by a timestamp when the feed is being created
+  (e.g. ``2020-03-28T14-45-08.237134``)
+
+* ``%(batch_id)d`` - gets replaced by the sequence number of the batch.
+
+  Use :ref:`printf-style string formatting <python:old-string-formatting>` to
+  alter the number format. For example, to make the batch ID a 5-digit
+  number by introducing leading zeroes as needed, use ``%(batch_id)05d``
+  (e.g. ``3`` becomes ``00003``, ``123`` becomes ``00123``).
+
+For instance, if your settings include::
+
+    FEED_EXPORT_BATCH_ITEM_COUNT=100
+
+And your :command:`crawl` command line is::
+
+    scrapy crawl spidername -o 'dirname/%(batch_id)d-filename%(batch_time)s.json'
+
+The command line above can generate a directory tree like::
+
+->projectname
+-->dirname
+--->1-filename2020-03-28T14-45-08.237134.json
+--->2-filename2020-03-28T14-45-09.148903.json
+--->3-filename2020-03-28T14-45-10.046092.json
+
+Where the first and second files contain exactly 100 items. The last one contains
+100 items or fewer.
diff --git a/scrapy/extensions/feedexport.py b/scrapy/extensions/feedexport.py
index 61dad8726a6..21177b1b0c6 100644
--- a/scrapy/extensions/feedexport.py
+++ b/scrapy/extensions/feedexport.py
@@ -6,6 +6,7 @@
 
 import logging
 import os
+import re
 import sys
 import warnings
 from datetime import datetime
@@ -180,14 +181,16 @@ def _store_in_thread(self, file):
 
 
 class _FeedSlot:
-    def __init__(self, file, exporter, storage, uri, format, store_empty):
+    def __init__(self, file, exporter, storage, uri, format, store_empty, batch_id, uri_template):
         self.file = file
         self.exporter = exporter
         self.storage = storage
         # feed params
-        self.uri = uri
+        self.batch_id = batch_id
         self.format = format
         self.store_empty = store_empty
+        self.uri_template = uri_template
+        self.uri = uri
         # flags
         self.itemcount = 0
         self._exporting = False
@@ -244,63 +247,112 @@ def __init__(self, crawler):
         for uri, feed in self.feeds.items():
             if not self._storage_supported(uri):
                 raise NotConfigured
+            if not self._settings_are_valid():
+                raise NotConfigured
             if not self._exporter_supported(feed['format']):
                 raise NotConfigured
 
     def open_spider(self, spider):
         for uri, feed in self.feeds.items():
-            uri = uri % self._get_uri_params(spider, feed['uri_params'])
-            storage = self._get_storage(uri)
-            file = storage.open(spider)
-            exporter = self._get_exporter(
-                file=file,
-                format=feed['format'],
-                fields_to_export=feed['fields'],
-                encoding=feed['encoding'],
-                indent=feed['indent'],
-            )
-            slot = _FeedSlot(file, exporter, storage, uri, feed['format'], feed['store_empty'])
-            self.slots.append(slot)
-            if slot.store_empty:
-                slot.start_exporting()
+            uri_params = self._get_uri_params(spider, feed['uri_params'])
+            self.slots.append(self._start_new_batch(
+                batch_id=1,
+                uri=uri % uri_params,
+                feed=feed,
+                spider=spider,
+                uri_template=uri,
+            ))
 
     def close_spider(self, spider):
         deferred_list = []
         for slot in self.slots:
-            if not slot.itemcount and not slot.store_empty:
-                # We need to call slot.storage.store nonetheless to get the file
-                # properly closed.
-                d = defer.maybeDeferred(slot.storage.store, slot.file)
-                deferred_list.append(d)
-                continue
-            slot.finish_exporting()
-            logfmt = "%s %%(format)s feed (%%(itemcount)d items) in: %%(uri)s"
-            log_args = {'format': slot.format,
-                        'itemcount': slot.itemcount,
-                        'uri': slot.uri}
-            d = defer.maybeDeferred(slot.storage.store, slot.file)
-
-            # Use `largs=log_args` to copy log_args into function's scope
-            # instead of using `log_args` from the outer scope
-            d.addCallback(
-                lambda _, largs=log_args: logger.info(
-                    logfmt % "Stored", largs, extra={'spider': spider}
-                )
-            )
-            d.addErrback(
-                lambda f, largs=log_args: logger.error(
-                    logfmt % "Error storing", largs,
-                    exc_info=failure_to_exc_info(f), extra={'spider': spider}
-                )
-            )
+            d = self._close_slot(slot, spider)
             deferred_list.append(d)
         return defer.DeferredList(deferred_list) if deferred_list else None
 
+    def _close_slot(self, slot, spider):
+        if not slot.itemcount and not slot.store_empty:
+            # We need to call slot.storage.store nonetheless to get the file
+            # properly closed.
+            return defer.maybeDeferred(slot.storage.store, slot.file)
+        slot.finish_exporting()
+        logfmt = "%s %%(format)s feed (%%(itemcount)d items) in: %%(uri)s"
+        log_args = {'format': slot.format,
+                    'itemcount': slot.itemcount,
+                    'uri': slot.uri}
+        d = defer.maybeDeferred(slot.storage.store, slot.file)
+
+        # Use `largs=log_args` to copy log_args into function's scope
+        # instead of using `log_args` from the outer scope
+        d.addCallback(
+            lambda _, largs=log_args: logger.info(
+                logfmt % "Stored", largs, extra={'spider': spider}
+            )
+        )
+        d.addErrback(
+            lambda f, largs=log_args: logger.error(
+                logfmt % "Error storing", largs,
+                exc_info=failure_to_exc_info(f), extra={'spider': spider}
+            )
+        )
+        return d
+
+    def _start_new_batch(self, batch_id, uri, feed, spider, uri_template):
+        """
+        Redirect the output data stream to a new file.
+        Execute multiple times if FEED_EXPORT_BATCH_ITEM_COUNT setting or FEEDS.batch_item_count is specified
+        :param batch_id: sequence number of current batch
+        :param uri: uri of the new batch to start
+        :param feed: dict with parameters of feed
+        :param spider: user spider
+        :param uri_template: template of uri which contains %(batch_time)s or %(batch_id)d to create new uri
+        """
+        storage = self._get_storage(uri)
+        file = storage.open(spider)
+        exporter = self._get_exporter(
+            file=file,
+            format=feed['format'],
+            fields_to_export=feed['fields'],
+            encoding=feed['encoding'],
+            indent=feed['indent'],
+        )
+        slot = _FeedSlot(
+            file=file,
+            exporter=exporter,
+            storage=storage,
+            uri=uri,
+            format=feed['format'],
+            store_empty=feed['store_empty'],
+            batch_id=batch_id,
+            uri_template=uri_template,
+        )
+        if slot.store_empty:
+            slot.start_exporting()
+        return slot
+
     def item_scraped(self, item, spider):
+        slots = []
         for slot in self.slots:
             slot.start_exporting()
             slot.exporter.export_item(item)
             slot.itemcount += 1
+            # create new slot for each slot with itemcount == FEED_EXPORT_BATCH_ITEM_COUNT and close the old one
+            if (
+                self.feeds[slot.uri_template]['batch_item_count']
+                and slot.itemcount >= self.feeds[slot.uri_template]['batch_item_count']
+            ):
+                uri_params = self._get_uri_params(spider, self.feeds[slot.uri_template]['uri_params'], slot)
+                self._close_slot(slot, spider)
+                slots.append(self._start_new_batch(
+                    batch_id=slot.batch_id + 1,
+                    uri=slot.uri_template % uri_params,
+                    feed=self.feeds[slot.uri_template],
+                    spider=spider,
+                    uri_template=slot.uri_template,
+                ))
+            else:
+                slots.append(slot)
+        self.slots = slots
 
     def _load_components(self, setting_prefix):
         conf = without_none_values(self.settings.getwithbase(setting_prefix))
@@ -317,6 +369,22 @@ def _exporter_supported(self, format):
             return True
         logger.error("Unknown feed format: %(format)s", {'format': format})
 
+    def _settings_are_valid(self):
+        """
+        If FEED_EXPORT_BATCH_ITEM_COUNT setting or FEEDS.batch_item_count is specified uri has to contain
+        %(batch_time)s or %(batch_id)d to distinguish different files of partial output
+        """
+        for uri_template, values in self.feeds.items():
+            if values['batch_item_count'] and not re.search(r'%\(batch_time\)s|%\(batch_id\)', uri_template):
+                logger.error(
+                    '%(batch_time)s or %(batch_id)d must be in the feed URI ({}) if FEED_EXPORT_BATCH_ITEM_COUNT '
+                    'setting or FEEDS.batch_item_count is specified and greater than 0. For more info see: '
+                    'https://docs.scrapy.org/en/latest/topics/feed-exports.html#feed-export-batch-item-count'
+                    ''.format(uri_template)
+                )
+                return False
+        return True
+
     def _storage_supported(self, uri):
         scheme = urlparse(uri).scheme
         if scheme in self.storages:
@@ -342,12 +410,14 @@ def _get_exporter(self, file, format, *args, **kwargs):
     def _get_storage(self, uri):
         return self._get_instance(self.storages[urlparse(uri).scheme], uri)
 
-    def _get_uri_params(self, spider, uri_params):
+    def _get_uri_params(self, spider, uri_params, slot=None):
         params = {}
         for k in dir(spider):
             params[k] = getattr(spider, k)
-        ts = datetime.utcnow().replace(microsecond=0).isoformat().replace(':', '-')
-        params['time'] = ts
+        utc_now = datetime.utcnow()
+        params['time'] = utc_now.replace(microsecond=0).isoformat().replace(':', '-')
+        params['batch_time'] = utc_now.isoformat().replace(':', '-')
+        params['batch_id'] = slot.batch_id + 1 if slot is not None else 1
         uripar_function = load_object(uri_params) if uri_params else lambda x, y: None
         uripar_function(params, spider)
         return params
diff --git a/scrapy/settings/default_settings.py b/scrapy/settings/default_settings.py
index 077317c815b..0016bbe1b06 100644
--- a/scrapy/settings/default_settings.py
+++ b/scrapy/settings/default_settings.py
@@ -146,6 +146,7 @@
     's3': 'scrapy.extensions.feedexport.S3FeedStorage',
     'ftp': 'scrapy.extensions.feedexport.FTPFeedStorage',
 }
+FEED_EXPORT_BATCH_ITEM_COUNT = 0
 FEED_EXPORTERS = {}
 FEED_EXPORTERS_BASE = {
     'json': 'scrapy.exporters.JsonItemExporter',
diff --git a/scrapy/utils/conf.py b/scrapy/utils/conf.py
index 5921f82bf8b..64f9c824b50 100644
--- a/scrapy/utils/conf.py
+++ b/scrapy/utils/conf.py
@@ -113,6 +113,7 @@ def get_sources(use_closest=True):
 
 def feed_complete_default_values_from_settings(feed, settings):
     out = feed.copy()
+    out.setdefault("batch_item_count", settings.getint('FEED_EXPORT_BATCH_ITEM_COUNT'))
     out.setdefault("encoding", settings["FEED_EXPORT_ENCODING"])
     out.setdefault("fields", settings.getlist("FEED_EXPORT_FIELDS") or None)
     out.setdefault("store_empty", settings.getbool("FEED_STORE_EMPTY"))
diff --git a/tests/test_feedexport.py b/tests/test_feedexport.py
index e386442140a..4e0b867a417 100644
--- a/tests/test_feedexport.py
+++ b/tests/test_feedexport.py
@@ -6,6 +6,8 @@
 import string
 import tempfile
 import warnings
+from abc import ABC, abstractmethod
+from collections import defaultdict
 from io import BytesIO
 from logging import getLogger
 from pathlib import Path
@@ -24,8 +26,9 @@
 
 import scrapy
 from scrapy.crawler import CrawlerRunner
+from scrapy.exceptions import NotConfigured
 from scrapy.exporters import CsvItemExporter
-from scrapy.extensions.feedexport import (BlockingFeedStorage, FileFeedStorage, FTPFeedStorage,
+from scrapy.extensions.feedexport import (BlockingFeedStorage, FeedExporter, FileFeedStorage, FTPFeedStorage,
                                           IFeedStorage, S3FeedStorage, StdoutFeedStorage)
 from scrapy.settings import Settings
 from scrapy.utils.python import to_unicode
@@ -79,6 +82,7 @@ class FTPFeedStorageTest(unittest.TestCase):
     def get_test_spider(self, settings=None):
         class TestSpider(scrapy.Spider):
             name = 'test_spider'
+
         crawler = get_crawler(settings_dict=settings)
         spider = TestSpider.from_crawler(crawler)
         return spider
@@ -132,6 +136,7 @@ class BlockingFeedStorageTest(unittest.TestCase):
     def get_test_spider(self, settings=None):
         class TestSpider(scrapy.Spider):
             name = 'test_spider'
+
         crawler = get_crawler(settings_dict=settings)
         spider = TestSpider.from_crawler(crawler)
         return spider
@@ -433,23 +438,84 @@ def store(self, file):
         file.close()
 
 
-class FeedExportTest(unittest.TestCase):
+class FeedExportTestBase(ABC, unittest.TestCase):
+    __test__ = False
 
     class MyItem(scrapy.Item):
         foo = scrapy.Field()
         egg = scrapy.Field()
         baz = scrapy.Field()
 
+    def _random_temp_filename(self, inter_dir=''):
+        chars = [random.choice(ascii_letters + digits) for _ in range(15)]
+        filename = ''.join(chars)
+        return os.path.join(self.temp_dir, inter_dir, filename)
+
     def setUp(self):
         self.temp_dir = tempfile.mkdtemp()
 
     def tearDown(self):
         shutil.rmtree(self.temp_dir, ignore_errors=True)
 
-    def _random_temp_filename(self):
-        chars = [random.choice(ascii_letters + digits) for _ in range(15)]
-        filename = ''.join(chars)
-        return os.path.join(self.temp_dir, filename)
+    @defer.inlineCallbacks
+    def exported_data(self, items, settings):
+        """
+        Return exported data which a spider yielding ``items`` would return.
+        """
+
+        class TestSpider(scrapy.Spider):
+            name = 'testspider'
+
+            def parse(self, response):
+                for item in items:
+                    yield item
+
+        data = yield self.run_and_export(TestSpider, settings)
+        return data
+
+    @defer.inlineCallbacks
+    def exported_no_data(self, settings):
+        """
+        Return exported data which a spider yielding no ``items`` would return.
+        """
+
+        class TestSpider(scrapy.Spider):
+            name = 'testspider'
+
+            def parse(self, response):
+                pass
+
+        data = yield self.run_and_export(TestSpider, settings)
+        return data
+
+    @defer.inlineCallbacks
+    def assertExported(self, items, header, rows, settings=None, ordered=True):
+        yield self.assertExportedCsv(items, header, rows, settings, ordered)
+        yield self.assertExportedJsonLines(items, rows, settings)
+        yield self.assertExportedXml(items, rows, settings)
+        yield self.assertExportedPickle(items, rows, settings)
+        yield self.assertExportedMarshal(items, rows, settings)
+        yield self.assertExportedMultiple(items, rows, settings)
+
+    @abstractmethod
+    def run_and_export(self, spider_cls, settings):
+        pass
+
+    def _load_until_eof(self, data, load_func):
+        result = []
+        with tempfile.TemporaryFile() as temp:
+            temp.write(data)
+            temp.seek(0)
+            while True:
+                try:
+                    result.append(load_func(temp))
+                except EOFError:
+                    break
+        return result
+
+
+class FeedExportTest(FeedExportTestBase):
+    __test__ = True
 
     @defer.inlineCallbacks
     def run_and_export(self, spider_cls, settings):
@@ -484,35 +550,6 @@ def run_and_export(self, spider_cls, settings):
 
         return content
 
-    @defer.inlineCallbacks
-    def exported_data(self, items, settings):
-        """
-        Return exported data which a spider yielding ``items`` would return.
-        """
-        class TestSpider(scrapy.Spider):
-            name = 'testspider'
-
-            def parse(self, response):
-                for item in items:
-                    yield item
-
-        data = yield self.run_and_export(TestSpider, settings)
-        return data
-
-    @defer.inlineCallbacks
-    def exported_no_data(self, settings):
-        """
-        Return exported data which a spider yielding no ``items`` would return.
-        """
-        class TestSpider(scrapy.Spider):
-            name = 'testspider'
-
-            def parse(self, response):
-                pass
-
-        data = yield self.run_and_export(TestSpider, settings)
-        return data
-
     @defer.inlineCallbacks
     def assertExportedCsv(self, items, header, rows, settings=None, ordered=True):
         settings = settings or {}
@@ -578,18 +615,6 @@ def assertExportedMultiple(self, items, rows, settings=None):
         json_rows = json.loads(to_unicode(data['json']))
         self.assertEqual(rows, json_rows)
 
-    def _load_until_eof(self, data, load_func):
-        result = []
-        with tempfile.TemporaryFile() as temp:
-            temp.write(data)
-            temp.seek(0)
-            while True:
-                try:
-                    result.append(load_func(temp))
-                except EOFError:
-                    break
-        return result
-
     @defer.inlineCallbacks
     def assertExportedPickle(self, items, rows, settings=None):
         settings = settings or {}
@@ -618,15 +643,6 @@ def assertExportedMarshal(self, items, rows, settings=None):
         result = self._load_until_eof(data['marshal'], load_func=marshal.load)
         self.assertEqual(expected, result)
 
-    @defer.inlineCallbacks
-    def assertExported(self, items, header, rows, settings=None, ordered=True):
-        yield self.assertExportedCsv(items, header, rows, settings, ordered)
-        yield self.assertExportedJsonLines(items, rows, settings)
-        yield self.assertExportedXml(items, rows, settings)
-        yield self.assertExportedPickle(items, rows, settings)
-        yield self.assertExportedMarshal(items, rows, settings)
-        yield self.assertExportedMultiple(items, rows, settings)
-
     @defer.inlineCallbacks
     def test_export_items(self):
         # feed exporters use field names from Item
@@ -650,7 +666,7 @@ def test_export_no_items_not_store_empty(self):
                 },
             }
             data = yield self.exported_no_data(settings)
-            self.assertEqual(data[fmt], b'')
+            self.assertEqual(b'', data[fmt])
 
     @defer.inlineCallbacks
     def test_export_no_items_store_empty(self):
@@ -670,7 +686,7 @@ def test_export_no_items_store_empty(self):
                 'FEED_EXPORT_INDENT': None,
             }
             data = yield self.exported_no_data(settings)
-            self.assertEqual(data[fmt], expctd)
+            self.assertEqual(expctd, data[fmt])
 
     @defer.inlineCallbacks
     def test_export_no_items_multiple_feeds(self):
@@ -1088,3 +1104,371 @@ def test_multiple_feeds_failing_logs_blocking_feed_storage(self):
         print(log)
         for fmt in ['json', 'xml', 'csv']:
             self.assertIn('Error storing %s feed (2 items)' % fmt, str(log))
+
+
+class BatchDeliveriesTest(FeedExportTestBase):
+    __test__ = True
+    _file_mark = '_%(batch_time)s_#%(batch_id)02d_'
+
+    @defer.inlineCallbacks
+    def run_and_export(self, spider_cls, settings):
+        """ Run spider with specified settings; return exported data. """
+
+        FEEDS = settings.get('FEEDS') or {}
+        settings['FEEDS'] = {
+            urljoin('file:', file_path): feed
+            for file_path, feed in FEEDS.items()
+        }
+        content = defaultdict(list)
+        try:
+            with MockServer() as s:
+                runner = CrawlerRunner(Settings(settings))
+                spider_cls.start_urls = [s.url('/')]
+                yield runner.crawl(spider_cls)
+
+            for path, feed in FEEDS.items():
+                dir_name = os.path.dirname(path)
+                for file in sorted(os.listdir(dir_name)):
+                    with open(os.path.join(dir_name, file), 'rb') as f:
+                        data = f.read()
+                        content[feed['format']].append(data)
+        finally:
+            self.tearDown()
+        defer.returnValue(content)
+
+    @defer.inlineCallbacks
+    def assertExportedJsonLines(self, items, rows, settings=None):
+        settings = settings or {}
+        settings.update({
+            'FEEDS': {
+                os.path.join(self._random_temp_filename(), 'jl', self._file_mark): {'format': 'jl'},
+            },
+        })
+        batch_size = settings.getint('FEED_EXPORT_BATCH_ITEM_COUNT')
+        rows = [{k: v for k, v in row.items() if v} for row in rows]
+        data = yield self.exported_data(items, settings)
+        for batch in data['jl']:
+            got_batch = [json.loads(to_unicode(batch_item)) for batch_item in batch.splitlines()]
+            expected_batch, rows = rows[:batch_size], rows[batch_size:]
+            self.assertEqual(expected_batch, got_batch)
+
+    @defer.inlineCallbacks
+    def assertExportedCsv(self, items, header, rows, settings=None, ordered=True):
+        settings = settings or {}
+        settings.update({
+            'FEEDS': {
+                os.path.join(self._random_temp_filename(), 'csv', self._file_mark): {'format': 'csv'},
+            },
+        })
+        batch_size = settings.getint('FEED_EXPORT_BATCH_ITEM_COUNT')
+        data = yield self.exported_data(items, settings)
+        for batch in data['csv']:
+            got_batch = csv.DictReader(to_unicode(batch).splitlines())
+            self.assertEqual(list(header), got_batch.fieldnames)
+            expected_batch, rows = rows[:batch_size], rows[batch_size:]
+            self.assertEqual(expected_batch, list(got_batch))
+
+    @defer.inlineCallbacks
+    def assertExportedXml(self, items, rows, settings=None):
+        settings = settings or {}
+        settings.update({
+            'FEEDS': {
+                os.path.join(self._random_temp_filename(), 'xml', self._file_mark): {'format': 'xml'},
+            },
+        })
+        batch_size = settings.getint('FEED_EXPORT_BATCH_ITEM_COUNT')
+        rows = [{k: v for k, v in row.items() if v} for row in rows]
+        data = yield self.exported_data(items, settings)
+        for batch in data['xml']:
+            root = lxml.etree.fromstring(batch)
+            got_batch = [{e.tag: e.text for e in it} for it in root.findall('item')]
+            expected_batch, rows = rows[:batch_size], rows[batch_size:]
+            self.assertEqual(expected_batch, got_batch)
+
+    @defer.inlineCallbacks
+    def assertExportedMultiple(self, items, rows, settings=None):
+        settings = settings or {}
+        settings.update({
+            'FEEDS': {
+                os.path.join(self._random_temp_filename(), 'xml', self._file_mark): {'format': 'xml'},
+                os.path.join(self._random_temp_filename(), 'json', self._file_mark): {'format': 'json'},
+            },
+        })
+        batch_size = settings.getint('FEED_EXPORT_BATCH_ITEM_COUNT')
+        rows = [{k: v for k, v in row.items() if v} for row in rows]
+        data = yield self.exported_data(items, settings)
+        # XML
+        xml_rows = rows.copy()
+        for batch in data['xml']:
+            root = lxml.etree.fromstring(batch)
+            got_batch = [{e.tag: e.text for e in it} for it in root.findall('item')]
+            expected_batch, xml_rows = xml_rows[:batch_size], xml_rows[batch_size:]
+            self.assertEqual(expected_batch, got_batch)
+        # JSON
+        json_rows = rows.copy()
+        for batch in data['json']:
+            got_batch = json.loads(batch.decode('utf-8'))
+            expected_batch, json_rows = json_rows[:batch_size], json_rows[batch_size:]
+            self.assertEqual(expected_batch, got_batch)
+
+    @defer.inlineCallbacks
+    def assertExportedPickle(self, items, rows, settings=None):
+        settings = settings or {}
+        settings.update({
+            'FEEDS': {
+                os.path.join(self._random_temp_filename(), 'pickle', self._file_mark): {'format': 'pickle'},
+            },
+        })
+        batch_size = settings.getint('FEED_EXPORT_BATCH_ITEM_COUNT')
+        rows = [{k: v for k, v in row.items() if v} for row in rows]
+        data = yield self.exported_data(items, settings)
+        import pickle
+        for batch in data['pickle']:
+            got_batch = self._load_until_eof(batch, load_func=pickle.load)
+            expected_batch, rows = rows[:batch_size], rows[batch_size:]
+            self.assertEqual(expected_batch, got_batch)
+
+    @defer.inlineCallbacks
+    def assertExportedMarshal(self, items, rows, settings=None):
+        settings = settings or {}
+        settings.update({
+            'FEEDS': {
+                os.path.join(self._random_temp_filename(), 'marshal', self._file_mark): {'format': 'marshal'},
+            },
+        })
+        batch_size = settings.getint('FEED_EXPORT_BATCH_ITEM_COUNT')
+        rows = [{k: v for k, v in row.items() if v} for row in rows]
+        data = yield self.exported_data(items, settings)
+        import marshal
+        for batch in data['marshal']:
+            got_batch = self._load_until_eof(batch, load_func=marshal.load)
+            expected_batch, rows = rows[:batch_size], rows[batch_size:]
+            self.assertEqual(expected_batch, got_batch)
+
+    @defer.inlineCallbacks
+    def test_export_items(self):
+        """ Test partial deliveries in all supported formats """
+        items = [
+            self.MyItem({'foo': 'bar1', 'egg': 'spam1'}),
+            self.MyItem({'foo': 'bar2', 'egg': 'spam2', 'baz': 'quux2'}),
+            self.MyItem({'foo': 'bar3', 'baz': 'quux3'}),
+        ]
+        rows = [
+            {'egg': 'spam1', 'foo': 'bar1', 'baz': ''},
+            {'egg': 'spam2', 'foo': 'bar2', 'baz': 'quux2'},
+            {'foo': 'bar3', 'baz': 'quux3', 'egg': ''}
+        ]
+        settings = {
+            'FEED_EXPORT_BATCH_ITEM_COUNT': 2
+        }
+        header = self.MyItem.fields.keys()
+        yield self.assertExported(items, header, rows, settings=Settings(settings))
+
+    def test_wrong_path(self):
+        """ If path is without %(batch_time)s and %(batch_id) an exception must be raised """
+        settings = {
+            'FEEDS': {
+                self._random_temp_filename(): {'format': 'xml'},
+            },
+            'FEED_EXPORT_BATCH_ITEM_COUNT': 1
+        }
+        crawler = get_crawler(settings_dict=settings)
+        self.assertRaises(NotConfigured, FeedExporter, crawler)
+
+    @defer.inlineCallbacks
+    def test_export_no_items_not_store_empty(self):
+        for fmt in ('json', 'jsonlines', 'xml', 'csv'):
+            settings = {
+                'FEEDS': {
+                    os.path.join(self._random_temp_filename(), fmt, self._file_mark): {'format': fmt},
+                },
+                'FEED_EXPORT_BATCH_ITEM_COUNT': 1
+            }
+            data = yield self.exported_no_data(settings)
+            data = dict(data)
+            self.assertEqual(b'', data[fmt][0])
+
+    @defer.inlineCallbacks
+    def test_export_no_items_store_empty(self):
+        formats = (
+            ('json', b'[]'),
+            ('jsonlines', b''),
+            ('xml', b'<?xml version="1.0" encoding="utf-8"?>\n<items></items>'),
+            ('csv', b''),
+        )
+
+        for fmt, expctd in formats:
+            settings = {
+                'FEEDS': {
+                    os.path.join(self._random_temp_filename(), fmt, self._file_mark): {'format': fmt},
+                },
+                'FEED_STORE_EMPTY': True,
+                'FEED_EXPORT_INDENT': None,
+                'FEED_EXPORT_BATCH_ITEM_COUNT': 1,
+            }
+            data = yield self.exported_no_data(settings)
+            data = dict(data)
+            self.assertEqual(expctd, data[fmt][0])
+
+    @defer.inlineCallbacks
+    def test_export_multiple_configs(self):
+        items = [dict({'foo': u'FOO', 'bar': u'BAR'}), dict({'foo': u'FOO1', 'bar': u'BAR1'})]
+
+        formats = {
+            'json': ['[\n{"bar": "BAR"}\n]'.encode('utf-8'),
+                     '[\n{"bar": "BAR1"}\n]'.encode('utf-8')],
+            'xml': [
+                (
+                    '<?xml version="1.0" encoding="latin-1"?>\n'
+                    '<items>\n  <item>\n    <foo>FOO</foo>\n  </item>\n</items>'
+                ).encode('latin-1'),
+                (
+                    '<?xml version="1.0" encoding="latin-1"?>\n'
+                    '<items>\n  <item>\n    <foo>FOO1</foo>\n  </item>\n</items>'
+                ).encode('latin-1')
+            ],
+            'csv': ['foo,bar\r\nFOO,BAR\r\n'.encode('utf-8'),
+                    'foo,bar\r\nFOO1,BAR1\r\n'.encode('utf-8')],
+        }
+
+        settings = {
+            'FEEDS': {
+                os.path.join(self._random_temp_filename(), 'json', self._file_mark): {
+                    'format': 'json',
+                    'indent': 0,
+                    'fields': ['bar'],
+                    'encoding': 'utf-8',
+                },
+                os.path.join(self._random_temp_filename(), 'xml', self._file_mark): {
+                    'format': 'xml',
+                    'indent': 2,
+                    'fields': ['foo'],
+                    'encoding': 'latin-1',
+                },
+                os.path.join(self._random_temp_filename(), 'csv', self._file_mark): {
+                    'format': 'csv',
+                    'indent': None,
+                    'fields': ['foo', 'bar'],
+                    'encoding': 'utf-8',
+                },
+            },
+            'FEED_EXPORT_BATCH_ITEM_COUNT': 1,
+        }
+        data = yield self.exported_data(items, settings)
+        for fmt, expected in formats.items():
+            for expected_batch, got_batch in zip(expected, data[fmt]):
+                self.assertEqual(expected_batch, got_batch)
+
+    @defer.inlineCallbacks
+    def test_batch_item_count_feeds_setting(self):
+        items = [dict({'foo': u'FOO'}), dict({'foo': u'FOO1'})]
+        formats = {
+            'json': ['[{"foo": "FOO"}]'.encode('utf-8'),
+                     '[{"foo": "FOO1"}]'.encode('utf-8')],
+        }
+        settings = {
+            'FEEDS': {
+                os.path.join(self._random_temp_filename(), 'json', self._file_mark): {
+                    'format': 'json',
+                    'indent': None,
+                    'encoding': 'utf-8',
+                    'batch_item_count': 1,
+                },
+            },
+        }
+        data = yield self.exported_data(items, settings)
+        for fmt, expected in formats.items():
+            for expected_batch, got_batch in zip(expected, data[fmt]):
+                self.assertEqual(expected_batch, got_batch)
+
+    @defer.inlineCallbacks
+    def test_batch_path_differ(self):
+        """
+        Test that the name of all batch files differ from each other.
+        So %(batch_time)s replaced with the current date.
+        """
+        items = [
+            self.MyItem({'foo': 'bar1', 'egg': 'spam1'}),
+            self.MyItem({'foo': 'bar2', 'egg': 'spam2', 'baz': 'quux2'}),
+            self.MyItem({'foo': 'bar3', 'baz': 'quux3'}),
+        ]
+        settings = {
+            'FEEDS': {
+                os.path.join(self._random_temp_filename(), '%(batch_time)s'): {
+                    'format': 'json',
+                },
+            },
+            'FEED_EXPORT_BATCH_ITEM_COUNT': 1,
+        }
+        data = yield self.exported_data(items, settings)
+        self.assertEqual(len(items) + 1, len(data['json']))
+
+    @defer.inlineCallbacks
+    def test_s3_export(self):
+        """
+        Test export of items into s3 bucket.
+        S3_TEST_BUCKET_NAME, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY must be specified in tox.ini
+        to perform this test:
+        [testenv]
+        setenv =
+            AWS_SECRET_ACCESS_KEY = ABCD
+            AWS_ACCESS_KEY_ID = EFGH
+            S3_TEST_BUCKET_NAME = IJKL
+        """
+        try:
+            import boto3
+        except ImportError:
+            raise unittest.SkipTest("S3FeedStorage requires boto3")
+
+        assert_aws_environ()
+        s3_test_bucket_name = os.environ.get('S3_TEST_BUCKET_NAME')
+        access_key = os.environ.get('AWS_ACCESS_KEY_ID')
+        secret_key = os.environ.get('AWS_SECRET_ACCESS_KEY')
+        if not s3_test_bucket_name:
+            raise unittest.SkipTest("No S3 BUCKET available for testing")
+
+        chars = [random.choice(ascii_letters + digits) for _ in range(15)]
+        filename = ''.join(chars)
+        prefix = 'tmp/{filename}'.format(filename=filename)
+        s3_test_file_uri = 's3://{bucket_name}/{prefix}/%(batch_time)s.json'.format(
+            bucket_name=s3_test_bucket_name, prefix=prefix
+        )
+        storage = S3FeedStorage(s3_test_bucket_name, access_key, secret_key)
+        settings = Settings({
+            'FEEDS': {
+                s3_test_file_uri: {
+                    'format': 'json',
+                },
+            },
+            'FEED_EXPORT_BATCH_ITEM_COUNT': 1,
+        })
+        items = [
+            self.MyItem({'foo': 'bar1', 'egg': 'spam1'}),
+            self.MyItem({'foo': 'bar2', 'egg': 'spam2', 'baz': 'quux2'}),
+            self.MyItem({'foo': 'bar3', 'baz': 'quux3'}),
+        ]
+        verifyObject(IFeedStorage, storage)
+
+        class TestSpider(scrapy.Spider):
+            name = 'testspider'
+
+            def parse(self, response):
+                for item in items:
+                    yield item
+
+        s3 = boto3.resource('s3')
+        my_bucket = s3.Bucket(s3_test_bucket_name)
+        batch_size = settings.getint('FEED_EXPORT_BATCH_ITEM_COUNT')
+
+        with MockServer() as s:
+            runner = CrawlerRunner(Settings(settings))
+            TestSpider.start_urls = [s.url('/')]
+            yield runner.crawl(TestSpider)
+
+        for file_uri in my_bucket.objects.filter(Prefix=prefix):
+            content = get_s3_content_and_delete(s3_test_bucket_name, file_uri.key)
+            if not content and not items:
+                break
+            content = json.loads(content.decode('utf-8'))
+            expected_batch, items = items[:batch_size], items[batch_size:]
+            self.assertEqual(expected_batch, content)
diff --git a/tests/test_utils_conf.py b/tests/test_utils_conf.py
index e5d3ef582c2..f3ef3612741 100644
--- a/tests/test_utils_conf.py
+++ b/tests/test_utils_conf.py
@@ -149,6 +149,7 @@ def test_feed_complete_default_values_from_settings_empty(self):
             "FEED_EXPORT_INDENT": 42,
             "FEED_STORE_EMPTY": True,
             "FEED_URI_PARAMS": (1, 2, 3, 4),
+            "FEED_EXPORT_BATCH_ITEM_COUNT": 2,
         })
         new_feed = feed_complete_default_values_from_settings(feed, settings)
         self.assertEqual(new_feed, {
@@ -157,6 +158,7 @@ def test_feed_complete_default_values_from_settings_empty(self):
             "indent": 42,
             "store_empty": True,
             "uri_params": (1, 2, 3, 4),
+            "batch_item_count": 2,
         })
 
     def test_feed_complete_default_values_from_settings_non_empty(self):
@@ -169,6 +171,7 @@ def test_feed_complete_default_values_from_settings_non_empty(self):
             "FEED_EXPORT_FIELDS": ["f1", "f2", "f3"],
             "FEED_EXPORT_INDENT": 42,
             "FEED_STORE_EMPTY": True,
+            "FEED_EXPORT_BATCH_ITEM_COUNT": 2,
         })
         new_feed = feed_complete_default_values_from_settings(feed, settings)
         self.assertEqual(new_feed, {
@@ -177,6 +180,7 @@ def test_feed_complete_default_values_from_settings_non_empty(self):
             "indent": 42,
             "store_empty": True,
             "uri_params": None,
+            "batch_item_count": 2,
         })
 
 
diff --git a/tox.ini b/tox.ini
index 27d21ade2f0..ccbcfeaef37 100644
--- a/tox.ini
+++ b/tox.ini
@@ -12,6 +12,7 @@ deps =
     -ctests/constraints.txt
     -rtests/requirements-py3.txt
     # Extras
+    boto3>=1.13.0
     botocore>=1.3.23
     Pillow>=3.4.2
 passenv =


Code review comments:
diff_hunk: @@ -461,7 +461,7 @@ For instance, if your settings include::
 
 And your :command:`crawl` command line is::
 
-    scrapy crawl spidername -o dirname/%(batch_id)d-filename%(batch_time)s.json
+    scrapy crawl spidername -o 'dirname/%(batch_id)d-filename%(batch_time)s.json'
path: docs/topics/feed-exports.rst
text: @user1:
In Windows `cmd.exe` single quotes won’t work. See https://docs.scrapy.org/en/latest/intro/tutorial.html#extracting-data