Skip to content

Console writer

Module to define behaviour to write to console.

ConsoleWriter

Bases: Writer

Class to write data to console.

Source code in mkdocs/lakehouse_engine/packages/io/writers/console_writer.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class ConsoleWriter(Writer):
    """Class to write data to console."""

    _logger = LoggingHandler(__name__).get_logger()

    def __init__(
        self, output_spec: OutputSpec, df: DataFrame, data: OrderedDict | None
    ) -> None:
        """Construct ConsoleWriter instances.

        Args:
            output_spec: output specification
            df: dataframe to be written.
            data: list of all dfs generated on previous steps before writer.
        """
        super().__init__(output_spec, df, data)

    def write(self) -> None:
        """Write data to console."""
        self._output_spec.options = (
            self._output_spec.options if self._output_spec.options else {}
        )
        if not self._df.isStreaming:
            self._logger.info("Dataframe preview:")
            self._show_df(self._df, self._output_spec)
        else:
            self._logger.info("Stream Dataframe preview:")
            self._write_to_console_in_streaming_mode(
                self._df, self._output_spec, self._data
            )

    @staticmethod
    def _show_df(df: DataFrame, output_spec: OutputSpec) -> None:
        """Given a dataframe it applies Spark's show function to show it.

        Args:
            df: dataframe to be shown.
            output_spec: output specification.
        """
        options = output_spec.options or {}
        df.show(
            n=options.get("limit", 20),
            truncate=options.get("truncate", True),
            vertical=options.get("vertical", False),
        )

    @staticmethod
    def _show_streaming_df(output_spec: OutputSpec) -> Callable:
        """Define how to show a streaming df.

        Args:
            output_spec: output specification.

        Returns:
            A function to show df in the foreachBatch spark write method.
        """

        def inner(batch_df: DataFrame, batch_id: int) -> None:
            ExecEnv.get_for_each_batch_session(batch_df)
            ConsoleWriter._logger.info(f"Showing DF for batch {batch_id}")
            ConsoleWriter._show_df(batch_df, output_spec)

        return inner

    @staticmethod
    def _write_to_console_in_streaming_mode(
        df: DataFrame, output_spec: OutputSpec, data: OrderedDict | None
    ) -> None:
        """Write to console in streaming mode.

        Args:
            df: dataframe to write.
            output_spec: output specification.
            data: list of all dfs generated on previous steps before writer.
        """
        df_writer = df.writeStream.trigger(**Writer.get_streaming_trigger(output_spec))

        if (
            output_spec.streaming_micro_batch_transformers
            or output_spec.streaming_micro_batch_dq_processors
        ):
            stream_df = df_writer.foreachBatch(
                ConsoleWriter._write_transformed_micro_batch(output_spec, data)
            ).start()
        else:
            stream_df = df_writer.foreachBatch(
                ConsoleWriter._show_streaming_df(output_spec)
            ).start()

        if output_spec.streaming_await_termination:
            stream_df.awaitTermination(output_spec.streaming_await_termination_timeout)

    @staticmethod
    def _write_transformed_micro_batch(
        output_spec: OutputSpec, data: OrderedDict | None
    ) -> Callable:
        """Define how to write a streaming micro batch after transforming it.

        Args:
            output_spec: output specification.
            data: list of all dfs generated on previous steps before writer.

        Returns:
            A function to be executed in the foreachBatch spark write method.
        """

        def inner(batch_df: DataFrame, batch_id: int) -> None:
            ExecEnv.get_for_each_batch_session(batch_df)
            transformed_df = Writer.get_transformed_micro_batch(
                output_spec, batch_df, batch_id, data
            )

            if output_spec.streaming_micro_batch_dq_processors:
                transformed_df = Writer.run_micro_batch_dq_process(
                    transformed_df, output_spec.streaming_micro_batch_dq_processors
                )

            ConsoleWriter._show_df(transformed_df, output_spec)

        return inner

__init__(output_spec, df, data)

Construct ConsoleWriter instances.

Parameters:

Name Type Description Default
output_spec OutputSpec

output specification

 required
df DataFrame

dataframe to be written.

 required
data OrderedDict | None

list of all dfs generated on previous steps before writer.

 required
Source code in mkdocs/lakehouse_engine/packages/io/writers/console_writer.py 
19
20
21
22
23
24
25
26
27
28
29
def __init__(
    self, output_spec: OutputSpec, df: DataFrame, data: OrderedDict | None
) -> None:
    """Construct ConsoleWriter instances.

    Args:
        output_spec: output specification
        df: dataframe to be written.
        data: list of all dfs generated on previous steps before writer.
    """
    super().__init__(output_spec, df, data)

write()

Write data to console.

Source code in mkdocs/lakehouse_engine/packages/io/writers/console_writer.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
def write(self) -> None:
    """Write data to console."""
    self._output_spec.options = (
        self._output_spec.options if self._output_spec.options else {}
    )
    if not self._df.isStreaming:
        self._logger.info("Dataframe preview:")
        self._show_df(self._df, self._output_spec)
    else:
        self._logger.info("Stream Dataframe preview:")
        self._write_to_console_in_streaming_mode(
            self._df, self._output_spec, self._data
        )