def get_data_frame(client, program, start, stop, resolution=None):
data = {}<EOL>metadata = {}<EOL>c = client.execute(program, start=start, stop=stop, resolution=resolution)<EOL>for msg in c.stream():<EOL><INDENT>if isinstance(msg, messages.DataMessage):<EOL><INDENT>if msg.logical_timestamp_ms in data:<EOL><INDENT>data[msg.logical_timestamp_ms].update(msg.data)<EOL><DEDENT>else:<EOL><INDENT>data[msg.logical_timestamp_ms] = msg.data<EOL><DEDENT><DEDENT>elif isinstance(msg, messages.MetadataMessage):<EOL><INDENT>metadata[msg.tsid] = msg.properties<EOL><DEDENT><DEDENT>df = pandas.DataFrame.from_dict(data, orient='<STR_LIT:index>')<EOL>df.metadata = metadata<EOL>return df<EOL>
Executes the given program across the given time range (expressed in
    millisecond timestamps since Epoch), and returns a Pandas DataFrame
    containing the results, indexed by output timestamp.

    If the program contains multiple publish() calls, their outputs are merged
    into the returned DataFrame.