<SYSTEM_TASK:>
Solve the following problem using Python, implementing the functions described below, one line at a time
<END_TASK>
<USER_TASK:>
Description:
 def _rows_event_to_dict(e, stream):
 """ Convert RowsEvent to a dict Args: e (pymysqlreplication.row_event.RowsEvent):
 the event stream (pymysqlreplication.BinLogStreamReader):
 the stream that yields event Returns: dict: event's data as a dict """

    pk_cols = e.primary_key if isinstance(e.primary_key, (list, tuple)) \
        else (e.primary_key, )

    if isinstance(e, row_event.UpdateRowsEvent):
        sig = signals.rows_updated
        action = 'update'
        row_converter = _convert_update_row
    elif isinstance(e, row_event.WriteRowsEvent):
        sig = signals.rows_inserted
        action = 'insert'
        row_converter = _convert_write_row
    elif isinstance(e, row_event.DeleteRowsEvent):
        sig = signals.rows_deleted
        action = 'delete'
        row_converter = _convert_write_row
    else:
        assert False, 'Invalid binlog event'

    meta = {
        'time': e.timestamp,
        'log_pos': stream.log_pos,
        'log_file': stream.log_file,
        'schema': e.schema,
        'table': e.table,
        'action': action,
    }
    rows = list(map(row_converter, e.rows))
    for row in rows:
        row['keys'] = {k: row['values'][k] for k in pk_cols}
    return rows, meta