Running queries against multiple data sources in Presto

With the growing popularity of artificial intelligence (AI) comes the need to store vast amounts of data in more diverse formats than ever before. The data that is required to build an AI model might come from a combination of both structured databases (called data warehouses) and unstructured or semi-structured data stores that make up what is called a data lake. When data comes from both data warehouses and data lakes, it makes sense to use a data lakehouse to manage all the necessary data sources. A data lakehouse combines the query speed and data quality of a data warehouse with the flexibility and low storage cost of a data lake. Watsonx.data, the data store component of watsonx, is built on an open lakehouse architecture.

As with any data source, a query engine is required to interpret requests, access and manipulate the stored data, and return the results. The Presto query engine is a great choice for data lakehouse analytics due to its efficient processing of data at scale. You can read more about the basic concepts underpinning Presto in this article.

In this tutorial, we set up a miniature version of a data lakehouse and configure a Presto cluster to submit a query across a data lakehouse.

Prerequisites

We will be using the Docker CLI tools to create containers for each element of our cluster and lakehouse. We recommend Podman, which is a rootless, and therefore more secure, drop-in replacement for Docker. Install Podman and ensure that you successfully alias podman to docker by running the below command.

$ docker run hello-world

You should see output similar to the following:

!... Hello Podman World ...!

         .--"--.
       / -     - \
      / (O)   (O) \
   ~~~| -=(,Y,)=- |
    .---. /`  \   |~~
 ~/  o  o \~~~~.----. ~~
  | =(X)= |~  / (O (O) \
   ~~~~~~~  ~| =(Y_)=-  |
  ~~~~    ~~~|   U      |~~

Project:   https://github.com/containers/podman
Website:   https://podman.io
Documents: https://docs.podman.io
Twitter:   @Podman_io

If you don't get this output, manually alias podman and try again:

$ alias podman=docker

Steps

At a high level, these are the steps to complete in this tutorial:

Create a Docker network
Set up a Presto cluster with 1 coordinator and 2 workers
Set up a structured data source - MySQL - and add data
Set up an unstructured data source - MongoDB - and add data
Connect the data sources to the Presto cluster
Query the data with the Presto CLI

Step 1. Create a Docker network

In this tutorial, we will be using a Docker container for each node of our Presto cluster and for each of our data sources. However, first, we will create a Docker network, which provides the infrastructure that is needed for all the containers to communicate with one another.

Open a terminal, and then run the following command to create a Docker network named presto_network:

$ docker network create presto_network

You can verify that the network was created and view specific details by running the following command:

$ docker network inspect presto_network

You should see output similar to the following output:

[
    {
        "Name": "presto_net",
        "Id": "78e7db7a109af1bff731b1af644c8ae2b259d3e5407f855aa271f4634e1dffe7",
        "Created": "2023-11-10T21:05:11.849902265Z",
        "Scope": "local",
        "Driver": "bridge",
        "EnableIPv6": false,
        "IPAM": {
            "Driver": "default",
            "Options": {},
            "Config": [
                {
                    "Subnet": "172.20.0.0/16",
                    "Gateway": "172.20.0.1"
                }
            ]
        },
        "Internal": false,
        "Attachable": false,
        "Ingress": false,
        "ConfigFrom": {
            "Network": ""
        },
        "ConfigOnly": false,
        "Containers": {},
        "Options": {},
        "Labels": {}
    }
]

The "Containers" key is empty right now because we have not created any containers or added them to our network yet. We will do that in the next step.

Step 2. Set up a Presto cluster

A Presto cluster is comprised of server nodes of two primary types: coordinators and workers. In this tutorial, we will create a cluster that includes 1 coordinator and 2 workers. Each node will use the same Docker image as its base, but with slightly different configuration to distinguish them.

First, we must pull the Presto image. In this tutorial, we are using the Presto version 0.282 image. You can also visit the PrestoDB DockerHub page to attempt this tutorial with a different version.

$ docker pull prestodb/presto:0.282

In order to run any presto server node, a few configuration files must be present at startup:

config.properties
jvm.config

We need to supply these files to our container at the location the node expects when starting the container. To do so, we must create these files on our local machine. Create a file named jvm.config and add the following code to the file. This file will be passed to each container in order to define parameters for that node's runtime environment.

-server
-Xmx2G
-XX:+UseG1GC
-XX:G1HeapRegionSize=32M
-XX:+UseGCOverheadLimit
-XX:+ExplicitGCInvokesConcurrent
-XX:+HeapDumpOnOutOfMemoryError
-XX:+ExitOnOutOfMemoryError
-Djdk.attach.allowAttachSelf=true

While the jvm.config will be the same for each node in our example, the config.properties file differs between nodes. We will create a separate properties file for each node. First, create a file named coordinator.properties and add the following code:

coordinator=true
node-scheduler.include-coordinator=false
http-server.http.port=8080
discovery-server.enabled=true
discovery.uri=http://localhost:8080
node.environment=test

Let's go over these properties line-by-line. The first line specifies that the node with this properties file will be a coordinator node. The second line indicates that the coordinator node will not be doing any work itself; it only performs coordinator duties. The third line gives the port by which all internal and external communication will happen for this node. The fourth line states that a discovery server will be used to register all nodes to the same cluster. The fifth line specifies the URI that each node will use to register itself with the cluster's discovery server. Finally, the last line defines the name of the enviroment our node will run in. All nodes in the same cluster must set the node.environment name to the same value.

Let's create another file named worker1.properties and add the following contents:

coordinator=false
http-server.http.port=8081
discovery.uri=http://coordinator:8080
node.environment=test

This file will be passed to the first worker as its config.properties file. This set of properties indicates that this node will not be a coordinator, sets a unique port for its communication, and gives the cluster discovery server URI, which includes the name of our coordinator container, so the worker node can register itself.

The file worker2.properties will be similar. Its contents only differs from that of worker1.properties in that a different communication port is assigned:

coordinator=false
http-server.http.port=8082
discovery.uri=http://coordinator:8080
node.environment=test

We will user Docker volumes in order to mount our locally created configuration files to the containers. Mounting a volume is a two-way operation: any files changed in one location will also be change in the other. This means that we will need a separate directory on our local machine for each node. In the following set of commands, we create a directory for each node and copy (or move) the relevant files into each.

$ mkdir coordinator-config
$ mv coordinator.properties coordinator-config/config.properties
$ cp jvm.config coordinator-config

$ mkdir worker1-config
$ mv worker1.properties worker1-config/config.properties
$ cp jvm.config worker1-config

$ mkdir worker2-config
$ mv worker2.properties worker2-config/config.properties
$ cp jvm.config worker2-config

We now have all the pieces that we need to start our containers. We'll first start the coordinator container.

$docker run -d -p 8080:8080 -v$ (pwd)/coordinator-config:/opt/presto-server/etc --net presto_network --name coordinator prestodb/presto:0.282

In this command, we are stating that we want to run the prestodb/presto:0.282 image. We also supply a human-readable name, "coordinator", and specify that it should be part of the presto_network that we created earlier. The -d option means that the container will start in the background, and the -p option maps a port on your local machine (the left side of the colon) to a port within the container (the right side of the colon) for proper communication routing.

Recall that we set port 8080 as the coordinator node's communication port in its configuration properties, so we use 8080 on the right side of the colon here. Finally, we must supply the expected configuration files by mounting the appropriate local directory to the expected location with the -v option. The path to the file on your local machine is on the left side of the colon, and the location to which this file should be mapped is on the right side of the colon. The latter value will always be /opt/presto-server/etc because that is the location the node expects.

You can verify that the container is up by running the following command:

$ docker ps

The output should show that the node container named "coordinator" has a status of "Up 'x' seconds" or something similar.

We will now repeat the process for each of our worker nodes, changing only the name, the port mapping, and the volume mapping:

$docker run -d -p 8081:8081 -v$ (pwd)/worker1-config:/opt/presto-server/etc --net presto_network --name worker1 prestodb/presto:0.282

$docker run -d -p 8082:8082 -v$ (pwd)/worker2-config:/opt/presto-server/etc --net presto_network --name worker2 prestodb/presto:0.282

If all nodes have successfully started, you now have a 3 node Presto cluster running on your system. Make sure that the coordinator node has started successfully by running the following command:

docker logs --tail 100 coordinator

If the Presto server is up and running properly, the last lines of the output would like the following:

2023-11-14T04:03:22.246Z        INFO    main    com.facebook.presto.storage.TempStorageManager  -- Loading temp storage local --
2023-11-14T04:03:22.251Z        INFO    main    com.facebook.presto.storage.TempStorageManager  -- Loaded temp storage local --
2023-11-14T04:03:22.256Z        INFO    main    com.facebook.presto.server.PrestoServer ======== SERVER STARTED ========

If you don't see any errors or the SERVER STARTED notice, wait a few minutes and check the logs again. Make sure all nodes have the SERVER STARTED notice before continuing.

Let's open the Presto CLI on the coordinator node to confirm everything has been set up. To do so, we will use docker exec to run a command in an already running container. The following command starts the presto-cli in interactive mode on the coordinator container.

$ docker exec -it coordinator presto-cli --server localhost:8080

The presto> prompt should appear in the terminal window after some time. We can run a simple command to see if the workers are communicating with the coordinator. This command outputs all the available catalogs, or data sources, that we have connected to our Presto cluster.

presto> SHOW CATALOGS;
Catalog
---------
 system
(1 row)

Query 20231110_225153_00004_bxsyi, FINISHED, 2 nodes
Splits: 36 total, 36 done (100.00%)
[Latency: client-side: 0:18, server-side: 0:17] [0 rows, 0B] [0 rows/s, 0B/s]

This output shows us that we have one catalog, system, and that 2 nodes were used to fetch this information as expected. It also gives some additional details such as the query ID and its status.

Recall that the coordinator node itself does not perform any work on queries in our example because we set the node-scheduler.include-coordinator property to false.

The system catalog is a built-in catalog for our setup. It provides information and metrics about the currently running Presto cluster. The system catalog, like every catalog, is made up of several schemas. Schemas are simply a logical way to organize tables. A catalog and a schema together define a set of tables that can be queried. We can run a query on the node table in the system catalog's runtime schema to get a high-level overview of our nodes.

presto> SELECT * FROM system.runtime.nodes;

   node_id    |        http_uri        | node_version  | coordinator | state
--------------+------------------------+---------------+-------------+--------
 4ace80a1fe24 | http://172.20.0.2:8080 | 0.282-fadfd95 | true        | active
 67941ce4fc7c | http://172.20.0.4:8082 | 0.282-fadfd95 | false       | active
 a99a1b831d5c | http://172.20.0.3:8081 | 0.282-fadfd95 | false       | active
(3 rows)

Query 20231110_231802_00007_bxsyi, FINISHED, 2 nodes
Splits: 17 total, 17 done (100.00%)
[Latency: client-side: 0:03, server-side: 0:03] [3 rows, 162B] [1 rows/s, 58B/s]

The information in the table should match up with the configuration options that we supplied to our nodes. Our cluster is not of much use yet, however, with just the system catalog. We want to connect other data sources as catalogs in order to be able to query them. We will do this in the next section.

Exit the presto-cli session on the coordinator with the exit command:

presto> exit;

Step 3. Set up a structured data source

We'll now set up our first data source that we will later connect to our Presto cluster: a MySQL database. MySQL is a structured data source, meaning it stores relational tables made up of data organized into rows and columns. We will use Docker to set up our MySQL database. Pull the MySQL image:

$ docker pull mysql

No additional setup is required for our simple cluster. We can run the container immediately, supplying the network and a human-readable name to more easily manage our containers:

$ docker -d run --net presto_network --name presto-mysql -e MYSQL_ROOT_PASSWORD=presto -d mysql

We have also specified an environment variable in this command, which represents the password that will be used when connecting to the MySQL shell. Let's run the shell now:

$ docker exec -it presto-mysql mysql -u root -h localhost -p

You will be prompted to enter the password from the docker run command: type presto and press enter. The MySQL shell starts and displays mysql> at the prompt line. From here, we will create a table in MySQL that be part of our underlying data in our data lake. We first need to create a new schema, called a database in MySQL, into which we will put our new table. Create the new schema and verify that the presto_mysql schema was created successfully.

mysql> CREATE DATABASE presto_mysql;
mysql> SHOW DATABASES;

Next we will create a new table, "authors", in this schema and add data into it with the following sequence of commands:

mysql> CREATE TABLE presto_mysql.authors (id bigint, author_name varchar(255));
mysql> INSERT INTO presto_mysql.authors VALUES (1, "Douglas Adams"), (2, "Sylvia Plath"), (3, "Stephen King");

The table we've created includes two columns: an id that will be unique across all authors and an author name. We can verify that our data was inserted correctly:

mysql> SELECT * FROM presto_mysql.authors;

+------+---------------+
| id   | author_name   |
+------+---------------+
|    1 | Douglas Adams |
|    2 | Sylvia Plath  |
|    3 | Stephen King  |
+------+---------------+
3 rows in set (0.00 sec)

As we can see, the data is organized in a relational format. Exit the shell with the exit; command.

Step 4. Set up an unstructured data source

We'll now set up our second data source for our data lake: a MongoDB instance. MongoDB stores unstructured data. In the case of MongoDB, data is stored as collections of documents that are made up of a series of key-value pairs. We will use Docker for this as well. Pull the MongoDB image:

$ docker pull mongo

Once again, no additional setup is required for our example. We can run the container immediately, supplying the network and a human-readable name:

$ docker run -d --net presto_network --name presto-mongo mongo

Let's run the MongoDB shell:

$ docker exec -it presto-mongo mongosh

The MongoDB shell starts and displays test> at the prompt line. From here, we can create a collection of documents that will be our underlying data. The following command creates a schema to use, called a db in MongoDB, and selects it as our working schema.

test> use presto_mongo;

This also has the effect of changing the shell prompt to presto_mongo. We can now create a collection, books, and insert data into it:

presto_mongo> db.createCollection("books")
presto_mongo> db.books.insertOne({"id": 1, "book_title": "The Hitchhiker's Guide to the Galaxy"})
presto_mongo> db.books.insertOne({"id": 2, "book_title": "The Bell Jar"})
presto_mongo> db.books.insertOne({"id": 3, "book_title": "The Alchemist"})
presto_mongo> db.books.insertOne({"id": 3, "book_title": "The Stand"})

The documents we've created each include two columns: an id that maps to a specific author, and a book title. We can verify that our data was inserted correctly:

presto_mongo> db.books.find()
[
  {
    _id: ObjectId("6552bd973ccdf5df2887acad"),
    id: 1,
    book_title: "The Hitchhiker's Guide to the Galaxy"
  },
  {
    _id: ObjectId("6552bd9b3ccdf5df2887acae"),
    id: 2,
    book_title: 'The Bell Jar'
  },
  {
    _id: ObjectId("6552bd9d3ccdf5df2887acaf"),
    id: 3,
    book_title: 'The Alchemist'
  },
  {
    _id: ObjectId("6552bda03ccdf5df2887acb0"),
    id: 3,
    book_title: 'The Stand'
  }
]

As we can see, the data looks much different from that of our structured database. Exit the shell with the exit command.

Step 5. Connect data sources to Presto

Now that our underlying data exists, we need to connect our data sources to Presto. Presto accomplishes this using the concept of a "connector". A connector access and manipulates the data in the underlying data source for use by Presto. Every catalog is associated with a specific connector, and catalogs are registered with Presto cluster nodes using another configuration file.

Let's connect our MySQL data source first. Create a properties file on your local machine called mysql.properties and add the following contents:

connector.name=mysql
connection-url=jdbc:mysql://presto-mysql:3306
connection-user=root
connection-password=presto

These properties assign a name to the connector (mysql) and supply details that the nodes need in order to access the underlying data. The connection URL gives the container name and the default MySQL server port.

The connector.name property is mandatory for every catalog configuration file.

Now we must add the file to each node's configuration directory that we created earlier. Create a directory catalog in each node's configuration folder and then copy the mysql.properties file into each:

mkdir -p coordinator-config/catalog
mkdir -p worker1-config/catalog
mkdir -p worker2-config/catalog

cp mysql.properties coordinator-config/catalog
cp mysql.properties worker1-config/catalog
cp mysql.properties worker2-config/catalog

Now we need to do the same for the MongoDB catalog. We will create another properties file, mongodb.properties, in our local catalog folders for each node. The file contents are as follows:

connector.name=mongodb
mongodb.seeds=presto-mongo:27017

These properties assign a name to the connector (mongodb) and supply the container name and default MongoDB server port in the seeds property.

Copy the file into the already-created catalog directories.

cp mongodb.properties coordinator-config/catalog
cp mongodb.properties worker1-config/catalog
cp mongodb.properties worker2-config/catalog

Only one additional step is required access our catalogs in the Presto cluster: restart each of our node containers so that they can pick up the new configuration files.

$ docker restart coordinator worker1 worker2

The restart process may take a few moments. Make sure the SERVER STARTED output exists in the logs before proceeding.

Step 6. Query the data with the Presto CLI

Now we can start the Presto CLI on the coordinator again to verify that we can access the new data.

$ docker exec -it coordinator presto-cli --server localhost:8080

The following command should now show two more catalogs than it did when we ran it before we connected our data sources.

presto> SHOW CATALOGS;
 Catalog
---------
 mongodb
 mysql
 system
(3 rows)

We can also view the data that we created in our data sources by supplying the catalog, schema, and table name to a SELECT query:

presto> SELECT * FROM mongodb.presto_mongo.books;
  id  |              book_title
------+--------------------------------------
    1 | The Hitchhiker's Guide to the Galaxy
    2 | The Bell Jar
    3 | The Alchemist
    3 | The Stand
(4 rows)

As you can see, the data is now displayed in row and column relational format, which is much different from how it was displayed when we ran the equivalent in mongosh. The Presto connectors are what enable this transformation to occur.

Finally, we can run a federated query across our data sources. A federated query is one that queries tables from multiple different datasets. In this query, we want to connect the authors to their books using the id column:

presto> SELECT * FROM mysql.presto_mysql.authors as TABLE_A JOIN mongodb.presto_mongo.books AS TABLE_B ON TABLE_A.id=TABLE_B.id;
 id |  author_name  | id |              book_title
----+---------------+----+--------------------------------------
  1 | Douglas Adams |  1 | The Hitchhiker's Guide to the Galaxy
  2 | Sylvia Plath  |  2 | The Bell Jar
  3 | Stephen King  |  3 | The Stand
  3 | Stephen King  |  3 | The Alchemist
(4 rows)

As you can see, the authors are matched with their book using the information from both our underyling data sources: the MySQL table and the MongoDB collection.

Summary and next steps

Congratulations! In this tutorial, you learned how easy it is to get started with a simple Presto cluster and connect disparate data sources to it. While we have used a very small data lake for demonstration purposes, Presto works efficiently even at petabyte-scale. This scalability is one of the many reasons that Presto is one of the query engines that powers watsonx.data.

If you need an enterprise-grade platform for querying vast amounts of diverse data with SQL, try out watsonx.data. You can also explore more articles and tutorials about watsonx to learn more.

Products

Languages

Technologies

All Events

External Resources