> ## Documentation Index
> Fetch the complete documentation index at: https://anaconda.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Run SQL query with pandas
export const Comments = ({children}) => {
return ;
};
Pandas has utility functions that make it one line to create a table, store it in a database, and later run queries against the data. This page shows how to run a SQL query against a self-hosted database from a Metaflow flow, transform the results in a dataframe, and write them back to the database.
To run the full example locally, [install MySQL](https://dev.mysql.com/doc/mysql-installation-excerpt/5.7/en/) and set up a database called `test`. This example uses a Python function defined in the script containing the flow to create the table, but you can set up the table any way you prefer to interact with the database.
The flow shows how to:
* Access data in a [pandas](https://pandas.pydata.org) dataframe by running a SQL query on a local database.
* This example uses a [MySQL](https://www.mysql.com/) database, but you could also store data in [PostgreSQL](https://www.postgresql.org).
* Make a transformation to the dataframe.
* Save the result to a separate table in the database.
```py title="sql_query_local.py" expandable theme={null}
from metaflow import FlowSpec, step, Parameter
from sqlalchemy import create_engine
import pandas as pd
class LocalQueryFlow(FlowSpec):
@step
def start(self):
self.next(self.extract)
@step
def extract(self):
QUERY = f"SELECT * FROM {table_name}"
self.result = pd.read_sql(QUERY, con=conn)
self.next(self.transform)
@step
def transform(self):
f = lambda x: x["feat_1"] + x["feat_2"]
self.result["feat_12"] = self.result.apply(f,
axis=1)
self.next(self.write)
@step
def write(self):
self.result.to_sql(name=f"{table_name}_updated",
con=conn,
if_exists="replace")
self.next(self.end)
@step
def end(self):
conn.close()
### Local database configuration ###
db_path = 'mysql://:@localhost/data'
table_name = 'data'
engine = create_engine(db_path, echo=False)
conn = engine.connect()
def create_table(db_path, table, conn):
# Create the dataset
dataset = pd.DataFrame({"id": [1, 2],
"feat_1": ["foo", "bar"],
"feat_2": ["fizz", "buzz"]})
try: # Write the contents to the local database
dataset.to_sql(table, con=conn)
except ValueError:
print(f"{table} at {db_path} doesn't exist.")
if __name__ == "__main__":
create_table(db_path, table_name, conn)
LocalQueryFlow()
```
Replace \ with your MySQL username.
Replace \ with your MySQL password.
```bash theme={null}
python sql_query_local.py run
```
```text theme={null}
Workflow starting (run-id 610):
[610/start/3177 (pid 73064)] Task is starting.
[610/start/3177 (pid 73064)] Task finished successfully.
[610/extract/3178 (pid 73069)] Task is starting.
[610/extract/3178 (pid 73069)] Task finished successfully.
[610/transform/3179 (pid 73073)] Task is starting.
[610/transform/3179 (pid 73073)] Task finished successfully.
[610/write/3180 (pid 73082)] Task is starting.
[610/write/3180 (pid 73082)] Task finished successfully.
[610/end/3181 (pid 73086)] Task is starting.
[610/end/3181 (pid 73086)] Task finished successfully.
Done!
```
Run the following in any script or notebook to access the contents of the dataframe that was stored as a flow artifact with `self.result`:
```python theme={null}
from metaflow import Flow
run = Flow('LocalQueryFlow').latest_run
run.data.result
```
```text theme={null}
index id feat_1 feat_2 feat_12
0 0 1 foo fizz foofizz
1 1 2 bar buzz barbuzz
```