diff --git a/pytype/file_utils.py b/pytype/file_utils.py
index dc6c99a04..0860b18f9 100644
--- a/pytype/file_utils.py
+++ b/pytype/file_utils.py
@@ -5,6 +5,7 @@
 import os
 import re
 import sys
+import typing
 
 from pytype.platform_utils import path_utils
 
@@ -157,3 +158,12 @@ def is_file_script(filename, directory=None):
       except UnicodeDecodeError:
         return False
       return re.fullmatch(r"#!.+python3?", line) is not None
+
+
+def merge_csvs(output: str, files: typing.List[str]) -> None:
+  """Merge contents of csv into one output csv file."""
+  with open(output, "w") as f:
+    for file in files:
+      with open(file, "r") as ifile:
+        f.write(ifile.read())
+      os.remove(file)
diff --git a/pytype/file_utils_test.py b/pytype/file_utils_test.py
index 4dd79cfd4..775a6c7e5 100644
--- a/pytype/file_utils_test.py
+++ b/pytype/file_utils_test.py
@@ -302,5 +302,22 @@ def test_expand_with_cwd(self):
       )
 
 
+class TestMergeCSVFiles(unittest.TestCase):
+
+  def test_merge(self):
+    with test_utils.Tempdir() as d:
+      p1 = d.create_file("a.csv", "a,b,c\n1,2,3\n4,5,6\n")
+      p2 = d.create_file("b.csv", "a,b,c\n7,8,9\n10,11,12\n")
+      m = d.create_file("merged.csv")
+      file_utils.merge_csvs(m, [p1, p2])
+      with open(m) as f:
+        self.assertEqual(
+          f.read(),
+          "a,b,c\n1,2,3\n4,5,6\na,b,c\n7,8,9\n10,11,12\n"
+        )
+      self.assertFalse(path_utils.isfile(p1))
+      self.assertFalse(path_utils.isfile(p2))
+
+
 if __name__ == "__main__":
   unittest.main()
diff --git a/pytype/tools/analyze_project/config.py b/pytype/tools/analyze_project/config.py
index 5269d6084..2c068d8bf 100644
--- a/pytype/tools/analyze_project/config.py
+++ b/pytype/tools/analyze_project/config.py
@@ -7,9 +7,9 @@
 import os
 import sys
 import textwrap
-from typing import Any
+from typing import Any, Optional
 
-from pytype import config as pytype_config
+from pytype import config as pytype_config, datatypes
 from pytype import file_utils
 from pytype import utils
 from pytype.platform_utils import path_utils
@@ -85,11 +85,21 @@ class Item:
 }
 
 
+def _get_pytype_flag(flag: str) -> Optional[pytype_config._Arg]:
+  return next((arg for arg in pytype_config.ALL_OPTIONS if arg.flag == flag), None)
+
+
+CUSTOM_FLAGS = list(filter(None, [
+    _get_pytype_flag('output-errors-csv'),
+]))
+
+
 # The missing fields will be filled in by generate_sample_config_or_die.
 def _pytype_single_items():
   """Args to pass through to pytype_single."""
   out = {}
-  flags = pytype_config.FEATURE_FLAGS + pytype_config.EXPERIMENTAL_FLAGS
+  flags = pytype_config.FEATURE_FLAGS + pytype_config.EXPERIMENTAL_FLAGS + CUSTOM_FLAGS
+
   for arg in flags:
     opt = arg.args[0]
     dest = arg.get('dest')
@@ -289,3 +299,7 @@ def read_config_file_or_die(filepath):
     else:
       logging.info('No config file found. Using default configuration.')
   return ret
+
+
+def add_custom_flags(o: datatypes.ParserWrapper):
+  pytype_config.add_options(o, CUSTOM_FLAGS)
diff --git a/pytype/tools/analyze_project/parse_args.py b/pytype/tools/analyze_project/parse_args.py
index 3262c4287..308980c83 100644
--- a/pytype/tools/analyze_project/parse_args.py
+++ b/pytype/tools/analyze_project/parse_args.py
@@ -77,6 +77,8 @@ def parse_args(self, argv):
     args = self.create_initial_args(file_config_names)
     self._parser.parse_args(argv, args)
     self.clean_args(args, file_config_names)
+    # output_errors_csv is dependent on report_errors, so we set it here.
+    args.report_errors = hasattr(args, 'output_errors_csv') or getattr(args, 'report_errors', True)
     self.postprocess(args)
     return args
 
@@ -157,6 +159,7 @@ def make_parser():
   wrapper = datatypes.ParserWrapper(parser)
   pytype_config.add_basic_options(wrapper)
   pytype_config.add_feature_flags(wrapper)
+  config.add_custom_flags(wrapper)
   return Parser(parser, pytype_single_args=wrapper.actions)
 
 
diff --git a/pytype/tools/analyze_project/pytype_runner.py b/pytype/tools/analyze_project/pytype_runner.py
index a2065c096..09c114424 100644
--- a/pytype/tools/analyze_project/pytype_runner.py
+++ b/pytype/tools/analyze_project/pytype_runner.py
@@ -2,6 +2,7 @@
 
 import collections
 from collections.abc import Iterable, Sequence
+import hashlib
 import importlib
 import itertools
 import logging
@@ -186,8 +187,15 @@ def __init__(self, conf, sorted_sources):
     self.custom_options = [
         (k, getattr(conf, k)) for k in set(conf.__slots__) - set(config.ITEMS)]
     self.keep_going = conf.keep_going
+    self.output_errors_csv = conf.output_errors_csv
     self.jobs = conf.jobs
 
+  def cleanup(self) -> None:
+    if self.output_errors_csv:
+      c = path_utils.relpath(path_utils.dirname(self.ninja_file))
+      files = [f"{c}/{hashlib.blake2b(i.encode(), digest_size=16).hexdigest()}.csv" for i in self.filenames]
+      file_utils.merge_csvs(f"{c}/{self.output_errors_csv}", files=files)
+
   def set_custom_options(self, flags_with_values, binary_flags, report_errors):
     """Merge self.custom_options into flags_with_values and binary_flags."""
     for dest, value in self.custom_options:
@@ -221,6 +229,10 @@ def get_pytype_command_for_ninja(self, report_errors):
         '--nofail',
     }
     self.set_custom_options(flags_with_values, binary_flags, report_errors)
+
+    if self.output_errors_csv:
+      flags_with_values['--output-errors-csv'] = '$out_path'
+
     # Order the flags so that ninja recognizes commands across runs.
     return (
         exe +
@@ -354,6 +366,10 @@ def write_build_statement(self, module, action, deps, imports, suffix):
                   deps=deps,
                   imports=escape_ninja_path(imports),
                   module=module.name))
+      if self.output_errors_csv:
+        f.write('  out_path = {path}.csv\n'.format(
+                path=escape_ninja_path(hashlib.blake2b(module.full_path.encode(), digest_size=16).hexdigest())
+                ))
     return output
 
   def setup_build(self):
@@ -423,4 +439,5 @@ def run(self):
     ret = self.build()
     if not ret:
       print('Success: no errors found')
+    self.cleanup()
     return ret
diff --git a/pytype/tools/analyze_project/pytype_runner_test.py b/pytype/tools/analyze_project/pytype_runner_test.py
index 011bbda6b..017b52ae7 100644
--- a/pytype/tools/analyze_project/pytype_runner_test.py
+++ b/pytype/tools/analyze_project/pytype_runner_test.py
@@ -3,6 +3,7 @@
 import collections
 from collections.abc import Sequence
 import dataclasses
+import hashlib
 import re
 
 from pytype import config as pytype_config
@@ -395,6 +396,14 @@ def test_custom_option_no_report_errors(self):
     options = self.get_options(args)
     self.assertTrue(options.precise_return)
 
+  def test_output_errors_csv_param(self):
+    custom_conf = self.parser.config_from_defaults()
+    custom_conf.output_errors_csv = 'foo.csv'
+    self.runner = make_runner([], [], custom_conf)
+    args = self.runner.get_pytype_command_for_ninja(report_errors=True)
+    options = self.get_options(args)
+    self.assertEqual(options.output_errors_csv, '$out_path')
+
 
 class TestGetModuleAction(TestBase):
   """Tests for PytypeRunner.get_module_action."""
@@ -545,8 +554,10 @@ def test_write(self):
 class TestNinjaBuildStatement(TestBase):
   """Tests for PytypeRunner.write_build_statement."""
 
-  def write_build_statement(self, *args, **kwargs):
+  def write_build_statement(self, *args, output_errors_csv: bool = False, **kwargs):
     conf = self.parser.config_from_defaults()
+    if output_errors_csv:
+      conf.output_errors_csv = 'foo.csv'
     with test_utils.Tempdir() as d:
       conf.output = d.path
       runner = make_runner([], [], conf)
@@ -639,6 +650,14 @@ def test_path_mismatch(self):
         path_utils.join('bar', 'baz.pyi'),
     )
 
+  def test_output_errors_csv(self):
+    _, _, build_statement = self.write_build_statement(
+        Module('', 'foo.py', 'foo'), Action.CHECK, set(), 'imports', '', output_errors_csv=True
+    )
+    f_hash = hashlib.blake2b(Module('', 'foo.py', 'foo').full_path.encode(), digest_size=16).hexdigest()
+    f_name = pytype_runner.escape_ninja_path(f"{f_hash}.csv")
+    self.assertIn(f'  out_path = {f_name}', build_statement)
+
 
 class TestNinjaBody(TestBase):
   """Test PytypeRunner.setup_build."""