Browse Source

HUE-7252 [importer] CSV with quoted new lines in fields are not handled propertly

Romain Rigaux 8 years ago
parent
commit
779dddb

+ 35 - 25
desktop/libs/indexer/src/indexer/file_format.py

@@ -18,6 +18,7 @@ import gzip
 import operator
 import operator
 import itertools
 import itertools
 import logging
 import logging
+import StringIO
 
 
 from django.utils.translation import ugettext as _
 from django.utils.translation import ugettext as _
 
 
@@ -79,6 +80,7 @@ def get_file_format_instance(file, format_=None):
 
 
   return (matches[0] if matches else get_format_types()[0]).get_instance(file_stream, format_)
   return (matches[0] if matches else get_format_types()[0]).get_instance(file_stream, format_)
 
 
+
 class FileFormat(object):
 class FileFormat(object):
   _name = None
   _name = None
   _description = None
   _description = None
@@ -164,6 +166,7 @@ class FileFormat(object):
 
 
     return obj
     return obj
 
 
+
 class GrokkedFormat(FileFormat):
 class GrokkedFormat(FileFormat):
   _grok = None
   _grok = None
   _customizable = False
   _customizable = False
@@ -185,6 +188,7 @@ class GrokkedFormat(FileFormat):
   def fields(self):
   def fields(self):
     return self._fields
     return self._fields
 
 
+
 class HueLogFormat(GrokkedFormat):
 class HueLogFormat(GrokkedFormat):
   _name = "hue"
   _name = "hue"
   _description = _("Hue Log File")
   _description = _("Hue Log File")
@@ -218,9 +222,11 @@ class HueLogFormat(GrokkedFormat):
       Field("protocol", "string")
       Field("protocol", "string")
     ]
     ]
 
 
+
 class GrokLineFormat(GrokkedFormat):
 class GrokLineFormat(GrokkedFormat):
   _parse_type = "grok_line"
   _parse_type = "grok_line"
 
 
+
 class ApacheCombinedFormat(GrokLineFormat):
 class ApacheCombinedFormat(GrokLineFormat):
   _name = "combined_apache"
   _name = "combined_apache"
   _description = _("Combined Apache Log File")
   _description = _("Combined Apache Log File")
@@ -243,6 +249,7 @@ class ApacheCombinedFormat(GrokLineFormat):
       Field("field_line", "text_en")
       Field("field_line", "text_en")
     ]
     ]
 
 
+
 class RubyLogFormat(GrokLineFormat):
 class RubyLogFormat(GrokLineFormat):
   _name = "ruby_log"
   _name = "ruby_log"
   _description = _("Ruby Log")
   _description = _("Ruby Log")
@@ -259,6 +266,7 @@ class RubyLogFormat(GrokLineFormat):
       Field("field_line", "text_en")
       Field("field_line", "text_en")
     ]
     ]
 
 
+
 class SyslogFormat(GrokLineFormat):
 class SyslogFormat(GrokLineFormat):
   _name = "syslog"
   _name = "syslog"
   _description = _("Syslog")
   _description = _("Syslog")
@@ -337,11 +345,23 @@ class CSVFormat(FileFormat):
 
 
     return valid
     return valid
 
 
+  @classmethod
+  def _get_sample(cls, file_stream):
+    encoding = i18n.get_site_encoding()
+
+    for reader in [TextFileReader, GzipFileReader]:
+      file_stream.seek(0)
+      sample_data, sample_lines = reader.readlines(file_stream, encoding)
+      file_stream.seek(0)
+
+      if sample_data is not None:
+        yield sample_data, sample_lines
+
   @classmethod
   @classmethod
   def _guess_from_file_stream(cls, file_stream):
   def _guess_from_file_stream(cls, file_stream):
-    for sample in  cls._get_sample(file_stream):
+    for sample_data, sample_lines in cls._get_sample(file_stream):
       try:
       try:
-        dialect, has_header = cls._guess_dialect(sample)
+        dialect, has_header = cls._guess_dialect(sample_data)
         delimiter = dialect.delimiter
         delimiter = dialect.delimiter
         line_terminator = dialect.lineterminator
         line_terminator = dialect.lineterminator
         quote_char = dialect.quotechar
         quote_char = dialect.quotechar
@@ -351,7 +371,7 @@ class CSVFormat(FileFormat):
           "line_terminator": line_terminator,
           "line_terminator": line_terminator,
           "quote_char": quote_char,
           "quote_char": quote_char,
           "has_header": has_header,
           "has_header": has_header,
-          "sample": sample
+          "sample": sample_data
         })
         })
       except Exception:
       except Exception:
         LOG.exception('Warning, cannot read the file format.')
         LOG.exception('Warning, cannot read the file format.')
@@ -359,21 +379,9 @@ class CSVFormat(FileFormat):
     # Guess dialect failed, fall back to defaults:
     # Guess dialect failed, fall back to defaults:
     return cls()
     return cls()
 
 
-  @classmethod
-  def _get_sample(cls, file_stream):
-    encoding = i18n.get_site_encoding()
-
-    for reader in [TextFileReader, GzipFileReader]:
-      file_stream.seek(0)
-      lines = reader.readlines(file_stream, encoding)
-      file_stream.seek(0)
-
-      if lines is not None:
-        yield '\n'.join(lines)
-
   @classmethod
   @classmethod
   def _from_format(cls, file_stream, format_):
   def _from_format(cls, file_stream, format_):
-    for sample in cls._get_sample(file_stream):
+    for sample_data, sample_lines in cls._get_sample(file_stream):
       try:
       try:
         delimiter = format_["fieldSeparator"].encode('utf-8')
         delimiter = format_["fieldSeparator"].encode('utf-8')
         line_terminator = format_["recordSeparator"].encode('utf-8')
         line_terminator = format_["recordSeparator"].encode('utf-8')
@@ -385,7 +393,7 @@ class CSVFormat(FileFormat):
           "line_terminator": line_terminator,
           "line_terminator": line_terminator,
           "quote_char": quote_char,
           "quote_char": quote_char,
           "has_header": has_header,
           "has_header": has_header,
-          "sample": sample
+          "sample": sample_data
         })
         })
       except Exception:
       except Exception:
         LOG.exception('Warning, cannot read the file format.')
         LOG.exception('Warning, cannot read the file format.')
@@ -461,7 +469,9 @@ class CSVFormat(FileFormat):
   def _get_sample_reader(self, sample):
   def _get_sample_reader(self, sample):
     if self.line_terminator != '\n':
     if self.line_terminator != '\n':
       sample = sample.replace('\n', '\\n')
       sample = sample.replace('\n', '\\n')
-    return csv.reader(sample.split(self.line_terminator), delimiter=self.delimiter, quotechar=self.quote_char)
+      return csv.reader(sample.split(self.line_terminator), delimiter=self.delimiter, quotechar=self.quote_char)
+    else:
+      return csv.reader(StringIO.StringIO(sample), delimiter=self.delimiter, quotechar=self.quote_char)
 
 
   def _guess_field_names(self, sample):
   def _guess_field_names(self, sample):
     reader = self._get_sample_reader(sample)
     reader = self._get_sample_reader(sample)
@@ -510,11 +520,11 @@ class GzipFileReader(object):
     try:
     try:
       data = gz.read(IMPORT_PEEK_SIZE)
       data = gz.read(IMPORT_PEEK_SIZE)
     except IOError:
     except IOError:
-      return None
+      return None, None
     try:
     try:
-      return data.splitlines()[:IMPORT_PEEK_NLINES]
+      return data, itertools.islice(csv.reader(StringIO.StringIO(data)), IMPORT_PEEK_NLINES)
     except UnicodeError:
     except UnicodeError:
-      return None
+      return None, None
 
 
 
 
 class TextFileReader(object):
 class TextFileReader(object):
@@ -524,9 +534,9 @@ class TextFileReader(object):
   def readlines(fileobj, encoding):
   def readlines(fileobj, encoding):
     try:
     try:
       data = fileobj.read(IMPORT_PEEK_SIZE)
       data = fileobj.read(IMPORT_PEEK_SIZE)
-      return data.splitlines()[:IMPORT_PEEK_NLINES]
+      return data, itertools.islice(csv.reader(StringIO.StringIO(data)), IMPORT_PEEK_NLINES)
     except UnicodeError:
     except UnicodeError:
-      return None
+      return None, None
 
 
 
 
 class HiveFormat(CSVFormat):
 class HiveFormat(CSVFormat):
@@ -559,7 +569,7 @@ class HiveFormat(CSVFormat):
 
 
   @classmethod
   @classmethod
   def get_instance(cls, file_stream, format_):
   def get_instance(cls, file_stream, format_):
-    sample = cls._get_sample(file_stream)
+    sample_data, sample_lines = cls._get_sample(file_stream)
 
 
     fields = []
     fields = []
 
 
@@ -574,6 +584,6 @@ class HiveFormat(CSVFormat):
       "line_terminator": '\n',
       "line_terminator": '\n',
       "quote_char": '"',
       "quote_char": '"',
       "has_header": False,
       "has_header": False,
-      "sample": sample,
+      "sample": sample_data,
       "fields": format_["fields"]
       "fields": format_["fields"]
     })
     })

+ 1 - 1
desktop/libs/indexer/src/indexer/indexers/morphline.py

@@ -126,7 +126,7 @@ class MorphlineIndexer(object):
 
 
   def guess_field_types(self, data):
   def guess_field_types(self, data):
     file_format = get_file_format_instance(data['file'], data['format'])
     file_format = get_file_format_instance(data['file'], data['format'])
-    return file_format.get_fields() if file_format else {'columns':[]}
+    return file_format.get_fields() if file_format else {'columns': []}
 
 
   # Breadth first ordering of fields
   # Breadth first ordering of fields
   def get_field_list(self, field_data):
   def get_field_list(self, field_data):