(
        file_handle, delimiter=',', mandatory_fields=None)
Generator for reading a CSV file.

    Args:
        file_handle: a file-like object containing the CSV content.
        delimiter: character used as a field separator, default: ','
        mandatory_fields: list of fields that must be present in the CSV header.

    Raises:
        RuntimeError: when there are missing fields.
        DataIngestionError: when there are issues with the data ingestion.
Generator for reading a CSV file.
Generator for reading a CSV file .
def read_and_validate_csv(
        file_handle, delimiter=',', mandatory_fields=None):
    """Generator for reading a CSV file.

    Args:
        file_handle: a file-like object containing the CSV content.
        delimiter: character used as a field separator, default: ','
        mandatory_fields: list of fields that must be present in the CSV header.

    Raises:
        RuntimeError: when there are missing fields.
        DataIngestionError: when there are issues with the data ingestion.
    """
    if not mandatory_fields:
        mandatory_fields = TIMESKETCH_FIELDS
    # Ensures delimiter is a string.
    if not isinstance(delimiter, six.text_type):
        delimiter = codecs.decode(delimiter, 'utf8')

    header_reader = pandas.read_csv(file_handle, sep=delimiter, nrows=0)
    _validate_csv_fields(mandatory_fields, header_reader)

    if hasattr(file_handle, 'seek'):
        file_handle.seek(0)

    try:
        reader = pandas.read_csv(
            file_handle, sep=delimiter, chunksize=DEFAULT_CHUNK_SIZE)
        for idx, chunk in enumerate(reader):
            skipped_rows = chunk[chunk['datetime'].isnull()]
            if not skipped_rows.empty:
                logger.warning(
                    '{0} rows skipped since they were missing a datetime field '
                    'or it was empty '.format(len(skipped_rows)))

            # Normalize datetime to ISO 8601 format if it's not the case.
            try:
                # Lines with unrecognized datetime format will result in "NaT"
                # (not available) as its value and the event row will be
                # dropped in the next line
                chunk['datetime'] = pandas.to_datetime(
                    chunk['datetime'], errors='coerce')
                num_chunk_rows = chunk.shape[0]
                chunk.dropna(subset=['datetime'], inplace=True)
                if len(chunk) < num_chunk_rows:
                    logger.warning(
                        '{0} rows dropped from Rows {1} to {2} due to invalid '
                        'datetime values'.format(
                            num_chunk_rows - len(chunk),
                            idx * reader.chunksize,
                            idx * reader.chunksize + num_chunk_rows))
                chunk['timestamp'] = chunk['datetime'].dt.strftime(
                    '%s%f').astype(int)
                chunk['datetime'] = chunk['datetime'].apply(
                    Timestamp.isoformat).astype(str)
            except ValueError:
                logger.warning(
                    'Rows {0} to {1} skipped due to malformed '
                    'datetime values '.format(
                        idx * reader.chunksize,
                        idx * reader.chunksize + chunk.shape[0]))
                continue
            if 'tag' in chunk:
                chunk['tag'] = chunk['tag'].apply(_parse_tag_field)

            for _, row in chunk.iterrows():
                _scrub_special_tags(row)
                # Remove all NAN values from the pandas.Series.
                row.dropna(inplace=True)
                yield row.to_dict()
    except (pandas.errors.EmptyDataError, pandas.errors.ParserError) as e:
        error_string = 'Unable to read file, with error: {0!s}'.format(e)
        logger.error(error_string)
        raise errors.DataIngestionError(error_string) from e
def read_and_validate_csv ( file_handle , delimiter = ',' , mandatory_fields = None ) : if not mandatory_fields : mandatory_fields = TIMESKETCH_FIELDS # Ensures delimiter is a string. if not isinstance ( delimiter , six . text_type ) : delimiter = codecs . decode ( delimiter , 'utf8' ) header_reader = pandas . read_csv ( file_handle , sep = delimiter , nrows = 0 ) _validate_csv_fields ( mandatory_fields , header_reader ) if hasattr ( file_handle , 'seek' ) : file_handle . seek ( 0 ) try : reader = pandas . read_csv ( file_handle , sep = delimiter , chunksize = DEFAULT_CHUNK_SIZE ) for idx , chunk in enumerate ( reader ) : skipped_rows = chunk [ chunk [ 'datetime' ] . isnull ( ) ] if not skipped_rows . empty : logger . warning ( '{0} rows skipped since they were missing a datetime field ' 'or it was empty ' . format ( len ( skipped_rows ) ) ) # Normalize datetime to ISO 8601 format if it's not the case. try : # Lines with unrecognized datetime format will result in "NaT" # (not available) as its value and the event row will be # dropped in the next line chunk [ 'datetime' ] = pandas . to_datetime ( chunk [ 'datetime' ] , errors = 'coerce' ) num_chunk_rows = chunk . shape [ 0 ] chunk . dropna ( subset = [ 'datetime' ] , inplace = True ) if len ( chunk ) < num_chunk_rows : logger . warning ( '{0} rows dropped from Rows {1} to {2} due to invalid ' 'datetime values' . format ( num_chunk_rows - len ( chunk ) , idx * reader . chunksize , idx * reader . chunksize + num_chunk_rows ) ) chunk [ 'timestamp' ] = chunk [ 'datetime' ] . dt . strftime ( '%s%f' ) . astype ( int ) chunk [ 'datetime' ] = chunk [ 'datetime' ] . apply ( Timestamp . isoformat ) . astype ( str ) except ValueError : logger . warning ( 'Rows {0} to {1} skipped due to malformed ' 'datetime values ' . format ( idx * reader . chunksize , idx * reader . chunksize + chunk . shape [ 0 ] ) ) continue if 'tag' in chunk : chunk [ 'tag' ] = chunk [ 'tag' ] . apply ( _parse_tag_field ) for _ , row in chunk . iterrows ( ) : _scrub_special_tags ( row ) # Remove all NAN values from the pandas.Series. row . dropna ( inplace = True ) yield row . to_dict ( ) except ( pandas . errors . EmptyDataError , pandas . errors . ParserError ) as e : error_string = 'Unable to read file, with error: {0!s}' . format ( e ) logger . error ( error_string ) raise errors . DataIngestionError ( error_string ) from e