Summary

Edit custom query mongodb in field text (pymongo) If not edit, auto add query 'collection.find()'

Edit connection to mongodb

Tests

Testing in df return to data transform

cc:

Summary

Add a cooler and more feature intensive code editor

• Switch current code editor to use Ace
• Add in line numbers and syntax highlighting
• Implement NextJS workaround with instructions for adding more functionality to the editor.

Tests

1. Typed words and ran commands using the new editor for Custom Code & Filter actions.

Create code for custom actions

2. Testing it on the Suggestions List code button

Able to edit from the suggestions and apply

3. Update the instructions in the export pipelines

Read only

cc: @johnson-mage

I have this problem whenever I try to create a new block (except for transformers and sensors for some reason), where jinja can't load the template. The template files are present in C:\Users\benja\anaconda3\envs\mage-test\Lib\site-packages\mage_ai\data_preparation\templates in the folders data_exporters, data_loaders etc... Which I think should be the path where they are loaded from

I'm on Windows 11 Browser is Chrome (however this shouldn't matter)

Error:

Traceback (most recent call last):

  File "C:\Users\benja\anaconda3\envs\mage-test\lib\site-packages\tornado\web.py", line 1702, in _execute
    result = method(*self.path_args, **self.path_kwargs)
  File "C:\Users\benja\anaconda3\envs\mage-test\lib\site-packages\mage_ai\server\api\blocks.py", line 96, in post
    upstream_block_uuids=payload.get('upstream_blocks', []),
  File "C:\Users\benja\anaconda3\envs\mage-test\lib\site-packages\mage_ai\data_preparation\models\block\__init__.py", line 338, in create
    pipeline_type=pipeline.type if pipeline is not None else None,
  File "C:\Users\benja\anaconda3\envs\mage-test\lib\site-packages\mage_ai\data_preparation\templates\template.py", line 82, in load_template
    pipeline_type=pipeline_type,
  File "C:\Users\benja\anaconda3\envs\mage-test\lib\site-packages\mage_ai\data_preparation\templates\template.py", line 61, in fetch_template_source
    template_source = __fetch_data_loader_templates(config, pipeline_type=pipeline_type)
  File "C:\Users\benja\anaconda3\envs\mage-test\lib\site-packages\mage_ai\data_preparation\templates\template.py", line 107, in __fetch_data_loader_templates
    template_env.get_template(template_path).render(
  File "C:\Users\benja\anaconda3\envs\mage-test\lib\site-packages\jinja2\environment.py", line 997, in get_template
    return self._load_template(name, globals)
  File "C:\Users\benja\anaconda3\envs\mage-test\lib\site-packages\jinja2\environment.py", line 958, in _load_template
    template = self.loader.load(self, name, self.make_globals(globals))
  File "C:\Users\benja\anaconda3\envs\mage-test\lib\site-packages\jinja2\loaders.py", line 125, in load
    source, filename, uptodate = self.get_source(environment, name)
  File "C:\Users\benja\anaconda3\envs\mage-test\lib\site-packages\jinja2\loaders.py", line 194, in get_source
    pieces = split_template_path(template)
  File "C:\Users\benja\anaconda3\envs\mage-test\lib\site-packages\jinja2\loaders.py", line 35, in split_template_path
    raise TemplateNotFound(template)
jinja2.exceptions.TemplateNotFound: data_loaders\default.jinja
data_loaders\default.jinja

Summary

Improve the column reports page

• Refactor cell and tables for undefined values
• Add progress bar to percentage metrics
• Add more quality metrics to the page.

Tests

Went on localhost

Long table

Short Table

cc: @johnson-mage

Note this has big changes to design as it removes the border from the cells for the datatable.

Summary

Display a table of high correlation values

• On the column detail visualization view, add table for "Columns with high correlation" next to the correlations heatmap, for all correlation values of abs(0.5) or more.

Tests

Table with a highly correlated column

Table with a low correlation

Table with no correlation

cc: @johnson-mage

Summary

Add code for API endpoints, DB models, and client library methods. The code for the models is a bit messy, so I'm going to try to improve that a bit, but it works for now.

TODO:

• [ ] Add unit tests
• [ ] Add iFrame

Tests

tested in local jupyter notebook

requests:

cc: @wangxiaoyou1993

Summary

• see above

Tests

Tested "primary", "secondary", and "danger" props. Sample data distribution displays correctly.

One-sided distributions also display correctly.

Added tooltips (cursor looks off in the GIF but it works correctly):

cc: @johnson-mage @dy46

Summary

Added a query parameter called "status" URI -- /api/pipeline_runs?_limit=30&_offset=0&status=failed Added a filter for filtering results by "status" in process_pipeline_runs( )

Tests

Needs further testing on all possible status values

cc: @tommydangerous @wangxiaoyou1993

I have a project running with:

• mage start
• mage version 0.3.4
• windows machine using WSL

I seem to be unable to create a new pipeline. I can press the new pipeline button on the UI and refresh the page so it appears, but then I get the error:

Traceback (most recent call last):

  File "/mnt/c/Users/path_to/venv/lib/python3.7/site-packages/tornado/web.py", line 1702, in _execute

    result = method(*self.path_args, **self.path_kwargs)

  File "/mnt/c/Users/path_to/venv/lib/python3.7/site-packages/mage_ai/server/server.py", line 111, in get

    pipeline = Pipeline.get(pipeline_uuid)

  File "/mnt/c/Users/path_to/venv/lib/python3.7/site-packages/mage_ai/data_preparation/models/pipeline.py", line 137, in get

    return Pipeline(uuid)

  File "/mnt/c/Users/path_to/venv/lib/python3.7/site-packages/mage_ai/data_preparation/models/pipeline.py", line 38, in __init__

    self.load_config_from_yaml()

  File "/mnt/c/Users/path_to/venv/lib/python3.7/site-packages/mage_ai/data_preparation/models/pipeline.py", line 247, in load_config_from_yaml

    self.load_config(self.get_config_from_yaml())

  File "/mnt/c/Users/path_to/venv/lib/python3.7/site-packages/mage_ai/data_preparation/models/pipeline.py", line 269, in load_config

    blocks = [build_shared_args_kwargs(c, Block) for c in self.block_configs]

TypeError: 'NoneType' object is not iterable

'NoneType' object is not iterable

Summary

Integrated new configuration into templates and clients.

Tests

Local testing was performed by querying data warehouses and databases using the new framework. Unit tests were updated.

cc: @wangxiaoyou1993

Summary

This implementation uses the correlation matrix formula to compute the VIFs. This change should speed up the computation by ~$p^2$, with $p$ the number of params.

However, the suggestion is now "less" informative, and just says that there is a problem collinearity with the returned columns.

The issue with the current implementation is that it is not column order invariant. This is particularly concerning in cases such as the test below, where $z$ and $x$ are functions of some of the other variables, but the other variables are removed because they are first in the ordering.

Rather than removing columns in case VIF>5, maybe provide users with the matrix of variance decomposition proportions which gives greater insight into the structure of the relationships between the variables.

@tommydangerous, as an idea, expand suggestions to allow for the inclusion of graphs. Then, you could include a heat map of variance decomposition proportions, and cast a spell of illumination on the correlation structure for users.

Tests

rng = np.random.default_rng()

t = rng.normal(loc=4., scale=5., size=1000)
u = rng.normal(loc=4., scale=5., size=1000)
v = rng.normal(loc=3., scale=9., size=1000)
w = rng.normal(loc=7., scale=1., size=1000)

x = 2.*u + 7.*v + rng.normal(loc=7., scale=1., size=1000)
z = w + 6.*v + rng.normal(loc=1., scale=6., size=1000)
df = pd.DataFrame(np.stack([t, u, v, w, x, z], axis=-1),
                  columns=["t", "u", "v", "w", "x", "z"])

rcc = RemoveCollinearColumns(df, df.dtypes, [])
print(rcc.evaluate())

Bumps json5 from 1.0.1 to 1.0.2.

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

I'm testing Mage to see if it can be a good fit for a project I'm working on. I already have a pipeline in notebooks, using Spark Structured Streaming. I tried to just copy the notebooks to mage and use scratchpad, and it works fine, but then it's not possible to run it as a pipeline. I then tried to use data loader, transformer and data exporter, but I get some strange errors.

The pipeline reads from a Kafka topic using spark structured streaming, so first I thought I could just wrap the spark read method in a decorator and it would work. So I created a streaming pipeline, but then I can't make the data loader in spark and have to use the kafka yaml file. So instead I tried to use batch pipeline. Here is a super simple example

from pyspark.sql import SparkSession

if 'data_loader' not in globals():
    from mage_ai.data_preparation.decorators import data_loader
if 'test' not in globals():
    from mage_ai.data_preparation.decorators import test

spark = SparkSession.builder.appName("foo").getOrCreate()


@data_loader
def load_data(*args, **kwargs):
    return (spark.readStream.format("kafka")
            .option("kafka.bootstrap.servers", "localhost:9093")
            .option("subscribe", "example")
            .option("startingOffsets", "earliest")
            .load())


@test
def test_output(df, *args) -> None:
    """
    Template code for testing the output of the block.
    """
    assert df is not None, 'The output is undefined'

But I get the following error

---------------------------------------------------------------------------
AnalysisException                         Traceback (most recent call last)
Cell In[7], line 71
     68     else:
     69         return find(lambda val: val is not None, output)
---> 71 df = execute_custom_code()
     73 # Post processing code below (source: output_display.py)
     76 def __custom_output():

Cell In[7], line 55, in execute_custom_code()
     50     block.run_upstream_blocks()
     52 global_vars = {'env': 'dev', 'execution_date': datetime.datetime(2023, 1, 2, 21, 0, 21, 934826), 'event': {}} or dict()
---> 55 block_output = block.execute_sync(
     56     custom_code=code,
     57     global_vars=global_vars,
     58     analyze_outputs=True,
     59     update_status=True,
     60     test_execution=True,
     61 )
     62 if False:
     63     block.run_tests(custom_code=code, update_tests=False)

File /usr/local/lib/python3.10/site-packages/mage_ai/data_preparation/models/block/__init__.py:575, in Block.execute_sync(self, analyze_outputs, build_block_output_stdout, custom_code, execution_partition, global_vars, logger, run_all_blocks, test_execution, update_status, store_variables, verify_output, input_from_output, runtime_arguments, dynamic_block_index, dynamic_block_uuid, dynamic_upstream_block_uuids)
    568     if logger is not None:
    569         logger.exception(
    570             f'Failed to execute block {self.uuid}',
    571             block_type=self.type,
    572             block_uuid=self.uuid,
    573             error=err,
    574         )
--> 575     raise err
    576 finally:
    577     if update_status:

File /usr/local/lib/python3.10/site-packages/mage_ai/data_preparation/models/block/__init__.py:544, in Block.execute_sync(self, analyze_outputs, build_block_output_stdout, custom_code, execution_partition, global_vars, logger, run_all_blocks, test_execution, update_status, store_variables, verify_output, input_from_output, runtime_arguments, dynamic_block_index, dynamic_block_uuid, dynamic_upstream_block_uuids)
    542 if store_variables and self.pipeline.type != PipelineType.INTEGRATION:
    543     try:
--> 544         self.store_variables(
    545             variable_mapping,
    546             execution_partition=execution_partition,
    547             override_outputs=True,
    548             spark=(global_vars or dict()).get('spark'),
    549             dynamic_block_uuid=dynamic_block_uuid,
    550         )
    551     except ValueError as e:
    552         if str(e) == 'Circular reference detected':

File /usr/local/lib/python3.10/site-packages/mage_ai/data_preparation/models/block/__init__.py:1210, in Block.store_variables(self, variable_mapping, execution_partition, override, override_outputs, spark, dynamic_block_uuid)
   1208     if spark is not None and type(data) is pd.DataFrame:
   1209         data = spark.createDataFrame(data)
-> 1210     self.pipeline.variable_manager.add_variable(
   1211         self.pipeline.uuid,
   1212         uuid_to_use,
   1213         uuid,
   1214         data,
   1215         partition=execution_partition,
   1216     )
   1218 for uuid in removed_variables:
   1219     self.pipeline.variable_manager.delete_variable(
   1220         self.pipeline.uuid,
   1221         uuid_to_use,
   1222         uuid,
   1223     )

File /usr/local/lib/python3.10/site-packages/mage_ai/data_preparation/variable_manager.py:72, in VariableManager.add_variable(self, pipeline_uuid, block_uuid, variable_uuid, data, partition, variable_type)
     70 variable.delete()
     71 variable.variable_type = variable_type
---> 72 variable.write_data(data)

File /usr/local/lib/python3.10/site-packages/mage_ai/data_preparation/models/variable.py:153, in Variable.write_data(self, data)
    151     self.__write_parquet(data)
    152 elif self.variable_type == VariableType.SPARK_DATAFRAME:
--> 153     self.__write_spark_parquet(data)
    154 elif self.variable_type == VariableType.GEO_DATAFRAME:
    155     self.__write_geo_dataframe(data)

File /usr/local/lib/python3.10/site-packages/mage_ai/data_preparation/models/variable.py:279, in Variable.__write_spark_parquet(self, data)
    277 def __write_spark_parquet(self, data) -> None:
    278     (
--> 279         data.write
    280         .option('header', 'True')
    281         .mode('overwrite')
    282         .csv(self.variable_path)
    283     )

File /usr/local/lib/python3.10/site-packages/pyspark/sql/dataframe.py:338, in DataFrame.write(self)
    326 @property
    327 def write(self) -> DataFrameWriter:
    328     """
    329     Interface for saving the content of the non-streaming :class:`DataFrame` out into external
    330     storage.
   (...)
    336     :class:`DataFrameWriter`
    337     """
--> 338     return DataFrameWriter(self)

File /usr/local/lib/python3.10/site-packages/pyspark/sql/readwriter.py:731, in DataFrameWriter.__init__(self, df)
    729 self._df = df
    730 self._spark = df.sparkSession
--> 731 self._jwrite = df._jdf.write()

File /usr/local/lib/python3.10/site-packages/py4j/java_gateway.py:1321, in JavaMember.__call__(self, *args)
   1315 command = proto.CALL_COMMAND_NAME +\
   1316     self.command_header +\
   1317     args_command +\
   1318     proto.END_COMMAND_PART
   1320 answer = self.gateway_client.send_command(command)
-> 1321 return_value = get_return_value(
   1322     answer, self.gateway_client, self.target_id, self.name)
   1324 for temp_arg in temp_args:
   1325     temp_arg._detach()

File /usr/local/lib/python3.10/site-packages/pyspark/sql/utils.py:196, in capture_sql_exception.<locals>.deco(*a, **kw)
    192 converted = convert_exception(e.java_exception)
    193 if not isinstance(converted, UnknownException):
    194     # Hide where the exception came from that shows a non-Pythonic
    195     # JVM exception message.
--> 196     raise converted from None
    197 else:
    198     raise

AnalysisException: 'write' can not be called on streaming Dataset/DataFrame

Any idea why this isn't working?

Will it be possible in the future to run a pipeline without the data loader, transformer and data exporter?

Is your feature request related to a problem? Please describe. I have thousands of pipeline runs for a single pipeline. I want to filter and search for a specific one using values in the pipeline run’s variables.

Example: filter or search by execution date range

Is your feature request related to a problem? Please describe. We are planning to set up a local instance of Mage and are wondering if it is possible to support user authentication. We would like different users to be able to have their own unique account settings."

Describe the solution you'd like User authentication.

Describe alternatives you've considered Grouping on a shared account.

Additional context N/A

Summary

This PR enables the initial documentation setup for Mage, currently available at docs.mage.ai

🚀 Setup

Simply merge in this PR and your documentation will be connected!

👩‍💻 Development

Install the Mintlify CLI to preview the documentation changes locally. To install, use the following command

npm i mintlify -g

Run the following command at the root of your documentation (where mint.json is)

mintlify dev

😎 Publishing Changes

Changes will be deployed to production automatically after pushing to the default branch.

You can also preview changes using PRs, which generates a preview link of the docs.

Troubleshooting

• Mintlify dev isn't running - Run mintlify install it'll re-install dependencies.
• Mintlify dev is updating really slowly - Run mintlify clear to clear the cache.

Is your feature request related to a problem? Please describe. Yes, the problem is that in complex production environments, different tasks within a pipeline often require different programming languages and/or Python environments. This can make it difficult to manage and execute these tasks in an efficient and scalable way.

Describe the solution you'd like I would like the ability to define pipelines where tasks can be executed in separate Docker containers, potentially using multiple Docker images. This would allow for better isolation and management of the different environments and languages required for each task.

Describe alternatives you've considered One alternative that I have considered is using Argo to run tasks as Docker containers. However, using YAML to define pipelines can become unwieldy for larger projects and organizations. In the past, I have worked on production projects where we used Airflow with the KubernetesPodOperator, which provided a more scalable solution.

Additional context If Mage is intended as a replacement for Airflow, it is important that it support the most common setup used by mature organizations in production environments. This includes the ability to run each task in its own isolated environment using Docker containers.