pathway.io.fs package

pathway.io.fs.read(path, format, *, value_columns, mode='streaming', primary_key=None, csv_settings=None, types=None, json_field_paths=None, persistent_id=None, autocommit_duration_ms=None, debug_data=None)

Reads a table from one or several files with the specified format.

In case the folder is passed to the engine, the order in which files from the directory are processed is determined according to the modification time of files within this folder: they will be processed by ascending order of the modification time.

In case the format is “plaintext”, the table will consist of a single column data with each cell containing a single line from the file.

  • Parameters
    • path (str) – Path to the file or to the folder with files.
    • value_columns (OptionalListstr) – Names of the columns to be extracted from the files.
    • format (str) – Format of data to be read. Currently “csv”, “json” and “plaintext” formats are supported.
    • mode (str) – If set to “streaming”, the engine will wait for the new input files in the directory. Set it to “static”, it will only consider the available data and ingest all of it in one commit. Default value is “streaming”.
    • primary_key (OptionalListstr) – In case the table should have a primary key generated according toa subset of its columns, the set of columns should be specified in this field. Otherwise, the primary key will be generated randomly.
    • csv_settings (Optional[CsvParserSettings]) – Settings for the CSV parser. This parameter is used only in case the specified format is “csv”.
    • types (OptionalDictstr, PathwayType) – Dictionary containing the mapping between the columns and the data types (pw.Type) of the values of those columns. This parameter is optional, and if not provided the default type is pw.Type.ANY. Supported in “csv” and “json” formats.
    • json_field_paths (OptionalDictstr, str) – If the format is “json”, this field allows to map field namesinto path in the read json object. For the field which require such mapping, it should be given in the format <field_name>: <path to be mapped>, where the path to be mapped needs to be a JSON Pointer (RFC 6901).
    • persistent_id (Optionalint) – (unstable) An identifier, under which the state of the table will be persisted or None, if there is no need to persist the state of this table. When a program restarts, it restores the state for all input tables according to what was saved for their persistent_id. This way it’s possible to configure the start of computations from the moment they were terminated last time.
    • debug_data – Static data replacing original one when debug mode is active.
  • Returns
    The table read.
  • Return type
    Table

Example:

Consider you want to read a dataset, stored in the filesystem in a standard CSV format. The dataset contains data about pets and their owners.

For the sake of demonstration, you can prepare a small dataset by creating a CSV file via a unix command line tool:

printf "id,owner,pet\n1,Alice,dog\n2,Bob,dog\n3,Alice,cat\n4,Bob,dog" > dataset.csv

In order to read it into Pathway’s table, you can first do the import and then use the pw.io.fs.read method:

 import pathway as pw
 t = pw.io.fs.read("dataset.csv", format="csv", value_columns=["owner", "pet"])

Then, you can output the table in order to check the correctness of the read:

 pw.debug.compute_and_print(t, include_id=False)
owner pet
Alice dog
  Bob dog
Alice cat
  Bob dog

Similarly, we can do the same for JSON format.

First, we prepare a dataset:

printf "{\"id\":1,\"owner\":\"Alice\",\"pet\":\"dog\"}
{\"id\":2,\"owner\":\"Bob\",\"pet\":\"dog\"}
{\"id\":3,\"owner\":\"Bob\",\"pet\":\"cat\"}
{\"id\":4,\"owner\":\"Bob\",\"pet\":\"cat\"}" > dataset.jsonlines

And then, we use the method with the “json” format:

 t = pw.io.fs.read("dataset.jsonlines", format="json", value_columns=["owner", "pet"])

Now let’s try something different. Consider you have site access logs stored in a separate folder in several files. For the sake of simplicity, a log entry contains an access ID, an IP address and the login of the user.

A dataset, corresponding to the format described above can be generated, thanks to the following set of unix commands:

mkdir logs
printf "id,ip,login\n1,127.0.0.1,alice\n2,8.8.8.8,alice" > logs/part_1.csv
printf "id,ip,login\n3,8.8.8.8,bob\n4,127.0.0.1,alice" > logs/part_2.csv

Now, let’s see how you can use the connector in order to read the content of this directory into a table:

 t = pw.io.fs.read("logs/", format="csv", value_columns=["ip", "login"])

The only difference is that you specified the name of the directory instead of the file name, as opposed to what you had done in the previous example. It’s that simple!

Alternatively, we can do the same for the “json” variant:

The dataset creation would look as follows:

mkdir logs
printf "{\"id\":1,\"ip\":\"127.0.0.1\",\"login\":\"alice\"}
{\"id\":2,\"ip\":\"8.8.8.8\",\"login\":\"alice\"}" > logs/part_1.jsonlines
printf "{\"id\":3,\"ip\":\"8.8.8.8\",\"login\":\"bob\"}
{\"id\":4,\"ip\":\"127.0.0.1\",\"login\":\"alice\"}" > logs/part_2.jsonlines

While reading the data from logs folder can be expressed as:

 t = pw.io.fs.read("logs/", format="json", value_columns=["ip", "login"], mode="static")

But what if you are working with a real-time system, which generates logs all the time. The logs are being written and after a while they get into the log directory (this is also called “logs rotation”). Now, consider that there is a need to fetch the new files from this logs directory all the time. Would Pathway handle that? Sure!

The only difference would be in the usage of mode field. So the code snippet will look as follows:

 t = pw.io.fs.read("logs/", format="csv", value_columns=["ip", "login"], mode="streaming")

Or, for the “json” format case:

 t = pw.io.fs.read("logs/", format="json", value_columns=["ip", "login"], mode="streaming")

With this method, you obtain a table updated dynamically. The changes in the logs would incur changes in the Business-Intelligence ‘BI’-ready data, namely, in the tables you would like to output. To see how these changes are reported by Pathway, have a look at the “Streams of Updates and Snapshots” article.

Finally, a simple example for the plaintext format would look as follows:

 t = pw.io.fs.read("raw_dataset/lines.txt", format="plaintext")

pathway.io.fs.write(table, filename, format)

Writes table’s stream of updates to a file in the given format.

  • Parameters
    • table (Table) – Table to be written.
    • filename (str) – Path to the target output file.
    • format (str) – Format to use for data output. Currently, there are two supported formats: “json” and “csv”
  • Returns
    None

Example:

In this simple example you can see how table output works. First, import Pathway and create a table:

 import pathway as pw
 t = pw.debug.parse_to_table("age owner pet \n 1 10 Alice dog \n 2 9 Bob cat \n 3 8 Alice cat")

Consider you would want to output the stream of changes of this table in csv format. In order to do that you simply do:

 pw.io.fs.write(t, "table.csv", format="csv")

Now, let’s see what you have on the output:

cat table.csv
age,owner,pet,time,diff
10,"Alice","dog",0,1
9,"Bob","cat",0,1
8,"Alice","cat",0,1

The first three columns clearly represent the data columns you have. The column time represents the number of operations minibatch, in which each of the rows was read. In this example, since the data is static: you have 0. The diff is another element of this stream of updates. In this context, it is 1 because all three rows were read from the input. All in all, the extra information in time and diff columns - in this case - shows us that in the initial minibatch (time = 0), you have read three rows and all of them were added to the collection (diff = 1).

Alternatively, this data can be written in JSON format:

 pw.io.fs.write(t, "table.jsonlines", format="json")

Then, we can also check the output file by executing the command:

cat table.jsonlines
{"age":10,"owner":"Alice","pet":"dog","diff":1,"time":0}
{"age":9,"owner":"Bob","pet":"cat","diff":1,"time":0}
{"age":8,"owner":"Alice","pet":"cat","diff":1,"time":0}

As one can easily see, the values remain the same, while the format has changed to a plain JSON.