def METHOD_NAME(record, schema):
    """
    Converts TIMESTAMP, DATETIME, DATE, and TIME types to their respective
    types for parquet compatibility.
    :param record: record of data from beam pipeline
    :param schema: string schema dict.
    :return: record with converted TIMESTAMP, DATETIME, DATE, and/or TIME
    fields.
    """
    def _fix_primitive(record, field):
        """
        Converts the a value in the field in the record for parquet
        compatibility. This is mainly to consistently repeated types.
        :param record: record from data from beam pipeline.
        :param field: (bigquery.schema.SchemaField) to convert.
        """
        field_name = field['name']
        if field['type'] in ('TIMESTAMP', 'DATETIME'):
            record[field_name] = int(
                datetime_to_epoch_timestamp(record[field_name]))
        elif field['type'] == 'DATE':
            record[field_name] = int(date_to_epoch_date(record[field_name]))
        elif field['type'] == 'TIME':
            try:
                record[field_name] = datetime.datetime.strptime(
                    record[field_name], '%H:%M:%S').time()
            except ValueError:
                record[field_name] = datetime.datetime.strptime(
                    record[field_name], '%H:%M:%S.%f').time()
        return record[field_name]
    schema = schema['fields'] if isinstance(schema, dict) else schema
    for field in schema:
        field_name = field['name']
        if field['mode'] == 'REPEATED':
            fixed_array = []
            for value in record[field_name]:
                if field['type'] == 'RECORD':
                    record[field_name] = METHOD_NAME(
                        value, field['fields'])
                else:
                    fixed_array.append(_fix_primitive(value, field))
                    record[field_name] = fixed_array
        else:
            if field['type'] == 'RECORD':
                record[field_name] = METHOD_NAME(
                    record[field_name], field['fields'])
            else:
                record[field_name] = _fix_primitive(record, field)
    return [record]