Patent Description:
The object of the invention is to optimize the execution of a dataflow plan.

Methods, systems and computer program products are provided for automated runtime configuration for dataflows, such as to automatically select or adapt a runtime environment or resources to a dataflow plan prior to execution. Metadata generated for dataflows indicates dataflow information, such as numbers and types of sources, sinks and operations, and the amount of data being consumed, processed and written. Weighted dataflow plans are created from unweighted (e.g., codeless graphical) dataflow plans based on metadata. Weights that indicate operation complexity or resource consumption are generated (e.g., by a trained model) for data operations (e.g., features). A runtime environment and/or one or more resources to execute a dataflow plan is/are selected based on the weighted dataflow and/or a determined maximum flow. Preferences may be provided to influence weighting and runtime selections.

Further features and advantages of the invention, as well as the structure and operation of various embodiments, are described in detail below with reference to the accompanying drawings. It is noted that the invention is not limited to the specific embodiments described herein. Such embodiments are presented herein for illustrative purposes only. Additional embodiments will be apparent to persons skilled in the relevant art(s) based on the teachings contained herein.

Further, when a particular feature, structure, or characteristic is described in connection with an example embodiment, it is submitted that it is within the knowledge of one skilled in the art to effect such feature, structure, or characteristic in connection with other embodiments whether or not explicitly described.

In the discussion, unless otherwise stated, adjectives such as "substantially" and "about" modifying a condition or relationship characteristic of a feature or features of an example embodiment of the disclosure, are understood to mean that the condition or characteristic is defined to within tolerances that are acceptable for operation of the embodiment for an application for which it is intended.

A dataflow (e.g., an ETL or ELT process) may access or extract data from one or more sources, transform the data, and load the transformed data into one or more sinks. Automated runtime configuration for dataflows are desired to avoid guessing/ trial and error with respect to the selection of computing resources (e.g., processors, memory, storage, network resources) (also referred to as "runtime resources") to execute a data workflow. Such an automated runtime configuration for dataflows is desired to yield the correct number of resources, avoid disruptions, failures (e.g., resource allocation failure, out of memory), delays, and/or excessive costs to execute a workflow.

According to embodiments, an automated runtime configuration system for dataflows may intelligently select one or more (e.g., the most) suitable runtime for a given dataflow, such as by using Machine Learning (ML) and artificial intelligence (AI) models with dataflow telemetry, which permits data engineers to focus on data transformation logic without worrying about execution details.

Accordingly, embodiments are described herein for automated runtime configuration for dataflows. Such embodiments automatically may select or adapt a runtime environment or resources to a dataflow plan prior to execution, while eliminating guesswork and avoiding runtime failures and runtime scaling. Such embodiments may be configured in various ways.

For instance, <FIG> is a block diagram of an automated runtime configuration selection system <NUM> for dataflows, according to an example embodiment. System <NUM> presents one of many possible example implementations. System <NUM> may comprise any number of computing devices and/or servers, such as the example components illustrated in <FIG> and additional or alternative devices not expressly illustrated. Other types of computing environments involving automated runtime configuration for dataflows are also contemplated.

As shown in <FIG>, example system <NUM> includes one or more resource manager(s) <NUM>, resources <NUM>, one or more dataflow server(s) <NUM>, one or more computing device(s) <NUM>, one or more data source(s) <NUM>, one or more dataflow transformations <NUM>, and one or more data sink(s) <NUM>, which may be communicatively coupled together by one or more network(s) <NUM>. Any combination of computing device(s) <NUM>, resources <NUM>, dataflow server(s) <NUM>, data source(s) <NUM>, data transformations <NUM>, and/or data sink(s) <NUM> may be part of an organization (e.g., a business enterprise), part of a cloud-based third party entity, or form any other type of single- or multiple-entity configuration.

Network(s) <NUM> may include one or more of any of a local area network (LAN), a wide area network (WAN), a personal area network (PAN), a combination of communication networks, such as the Internet, and/or a virtual network. In example implementations, any one or more of resource manager(s) <NUM>, resources <NUM>, dataflow server(s) <NUM>, computing device(s) <NUM>, data source(s) <NUM>, dataflow transformation <NUM> and data sink(s) <NUM> may communicate over networks(s) <NUM> via one or more application programming interfaces (APIs), and/or according to other interfaces and/or techniques. Resource manager(s) <NUM>, resources <NUM>, dataflow server(s) <NUM>, computing device(s) <NUM>, data source(s) <NUM>, dataflow transformation <NUM> and data sink(s) <NUM> may each include at least one network interface that enables communications with each other. Examples of such a network interface, wired or wireless, include an IEEE <NUM> wireless LAN (WLAN) wireless interface, a Worldwide Interoperability for Microwave Access (Wi-MAX) interface, an Ethernet interface, a Universal Serial Bus (USB) interface, a cellular network interface, a Bluetooth™ interface, a near field communication (NFC) interface, etc. Further examples of network interfaces are described elsewhere herein.

Computing device(s) <NUM> may represent any number of computing devices, each of which may comprise any computing device utilized by one or more users. Computing device(s) <NUM> may comprise one or more applications, operating systems, virtual machines, storage devices, etc., that may be executed, hosted, and/or stored therein or by one or more other computing devices on or over network(s) <NUM>. In an example, computing device(s) <NUM> may access one or more server devices, resources and/or storage devices related to data transformations, such as, resources <NUM>, dataflow server(s) <NUM>, data source(s) <NUM>, dataflow transformation <NUM> and/or data sink(s) <NUM>. Computing device(s) <NUM> may each be any type of stationary or mobile computing device, including a mobile computer or mobile computing device (e.g., a Microsoft ® Surface® device, a personal digital assistant (PDA), a laptop computer, a notebook computer, a tablet computer such as an Apple iPad™, a netbook, etc.), a mobile phone, a wearable computing device, or other type of mobile device, or a stationary computing device such as a desktop computer or PC (personal computer), or a server. Computing device(s) <NUM> are not limited to physical machines, but may include other types of machines or nodes, such as a virtual machine. Computing device(s) <NUM> may each interface with dataflow server(s) <NUM> through APIs and/or by other mechanisms. Any number of program interfaces may coexist on computing device(s) <NUM>.

User(s) <NUM> includes one or more of any suitable type of user (e.g., individual users, family users, enterprise users, governmental users, etc.) at any number of locations. User(s) <NUM> may, for example, have a role within an enterprise (e.g., data engineer or another role). User(s) <NUM> may interact with dataflow server(s) <NUM> (e.g., via application <NUM>, computing device(s) <NUM> and network(s) <NUM>) to develop and execute dataflow plans. User(s) <NUM> may interact with application <NUM> displayed by computing device(s) <NUM>, for example, to provide input (e.g., dataflows, preferences), receive results, cause system <NUM> to perform functions (e.g., analyze and provide feedback for dataflows, execute dataflows, provide results, and/or request input). User preferences may include indications of, for example, whether execution is time-sensitive and/or cost sensitive. Preferences may become independent dataflow plan features and/or may influence metadata and/or weighting of other dataflow plan features.

Application <NUM> may comprise a dataflow design tool or other suitable tool/application implementable on one or more computing devices. Application <NUM> may be configured and used to create codeless (e.g., graphical) dataflows to process data. Application <NUM> may comprise, for example, a Web application provided by dataflow server(s) <NUM> or a locally executed application that interacts with dataflow server(s) <NUM> (e.g., through an application programming interface (API) or agent). User(s) <NUM> may use application <NUM> to create, edit, review and/or transform data stored on data source(s) <NUM> and data sink(s) <NUM>. User(s) <NUM> may layout and configure data sources, data sinks, operations, etc. on a dataflow design canvas provided in a GUI (graphical user interface) displayed by computing device(s) <NUM>. User(s) <NUM> may use application <NUM> to create and submit dataflow plans (e.g., dataflow plan(s) <NUM>) to, and receive feedback and/or results from, dataflow server(s) <NUM>.

Application <NUM> may provide a preview that user(s) <NUM> may select to receive a preliminary analysis to check for errors, receive an estimate of execution resources, time and/or cost, evaluate proposed changes (e.g., relative to errors, execution time and/or cost), and so on prior to submitting a dataflow plan for execution. A user may receive feedback while creating a dataflow, such as suggesting a left-join rather than a cross-join to reduce workload. Feedback and user preferences may allow users to edit and re-save dataflows with one or more improvements and/or preferences prior to submission to dataflow server(s) <NUM>.

In an example, Application <NUM> may provide a dataflow canvas in a graphical user interface displayed by computing device(s) <NUM>. A user may use application <NUM> (e.g., Microsoft® Azure Data Factory) to visually design data transformations in a logical flow (e.g., a graph with nodes and connecting edges) on the dataflow canvas without writing any code. Application <NUM> and/or dataflow server(s) <NUM> may realize mapping data flows created by user(s) <NUM> by providing translation of visual depictions to code, path optimization, execution planning (e.g., selecting an execution environment), scheduling and execution.

Dataflow plan(s) <NUM> may comprise one or more of any type of dataflow plan. In an example, a dataflow plan may comprise a codeless (e.g., graphical) specification indicating input data to be processed, how the input data is to be processed (e.g., applicable data transformations), and the data to be output. A dataflow (e.g., of dataflow(s) <NUM>) may indicate, for example, one or more data sources (e.g., data source(s) <NUM>), one or more dataflow transformations (e.g., dataflow transformations <NUM>) and one or more data sinks (e.g., data sink(s) <NUM>).

Resource manager(s) <NUM> manage resources <NUM>. Resource manager(s) <NUM> may be implemented on one or more computing devices (e.g., servers) to manage one or more types of computing resources (e.g., resources <NUM>). Managed computing resources (e.g., resources <NUM>) may comprise, for example, computing resources in a data center as well as computing resources in other locations. Types of computing resources may comprise, for example, processors, virtual machines (VMs), memory, network, storage, and/or other computing resources. Dataflow server(s) <NUM> (e.g., a dataflow scheduler) may communicate with resource manager(s) <NUM>, for example, to allocate and configure resources <NUM> to execute dataflow plans (e.g., dataflow plan(s) <NUM>). Resource manager(s) <NUM> may allocate and configure one or more servers and/or other resources, for example, to execute one or more dataflow plans (e.g., dataflow plan(s) <NUM>). In an example, resources allocated and configured to execute a dataflow plan may comprise a group of servers configured with an appropriate number of processor cores, amount of memory, amount of storage and network bandwidth.

Data source(s) <NUM>, dataflow transformations <NUM>, and data sink(s) <NUM> each store data. Data source(s) <NUM>, dataflow transformations <NUM>, and data sink(s) <NUM> (e.g., and storage resources that may be used for dataflow execution) may comprise any number and type of storage devices in any number of locations configured to store one or more types of data (e.g., information) in one or more formats. Data source(s) <NUM> may store data to be used as input to the execution of a dataflow plan. Data sink(s) <NUM> may store data to be generated by execution of a dataflow plan. Dataflow transformations <NUM> may store data transform algorithms that transform input source data into intermediate transformed data and, ultimately, sink (e.g., output) data during execution of a dataflow plan. Data source(s) <NUM>, dataflow transformations <NUM>, and/or data sink(s) <NUM> may have configurable settings (e.g., for data formats). In an example, user(s) <NUM> may configure data source(s) <NUM>, dataflow transformations <NUM>, and/or data sink(s) <NUM>, e.g., based on feature vectors that may be extracted from data and data transforms that may be applied to data during execution of a dataflow plan. Data source(s) <NUM>, dataflow transformations <NUM>, and data sink(s) <NUM> may comprise one or more databases that store structured data managed by a database engine. Objects of a database may comprise, for example, tables, records, indexes, schemas, metadata, etc..

Dataflow server(s) <NUM> may comprise any number of computing devices, servers, services, local processes, remote machines, web services, etc. for providing (e.g., managing) dataflow services to one or more users of computing devices (e.g., computing device(s) <NUM>). Dataflow server(s) <NUM> may provide an adaptive system to automatically size execution environments and computing resources therein to transform data for a dataflow plan, thereby providing a proactive approach rather than a reactive approach such as runtime scaling. Proactive and adaptive automatic execution environment selection before runtime reduces runtime dependency on resources. Dataflow server(s) <NUM> may run continuously, periodically, as a background service or other implementation.

Dataflow server(s) <NUM> may utilize one or more ML models. For example, one or more ML models may be trained based on known (e.g., historical) dataflow plans, execution environments and performance profiles to select execution environments for dataflow plans (e.g., ETL workloads), which may avoid (e.g., prevent) bottlenecks during runtime, e.g., along with avoiding delays and/or failures caused by bottlenecks). Automated execution environment (e.g., integration runtime) selection for a mapped data flow provided by dataflow server(s) <NUM> may increase runtime efficiency, reliability and cost-effectiveness for data processing pipelines. This is a proactive approach to allocate adequate resources, e.g., compared to a reactive approach (e.g., runtime scaling) that attempts to make adjustments during runtime for inadequacies detected in an execution environment. Execution environments and integration runtimes (IRs) may be referenced interchangeably.

Interface <NUM> provides an interface between computing device(s) <NUM> (e.g., application <NUM> executed thereon) and dataflow server(s) <NUM>. Interface <NUM> may comprise, for example, an API. Interface <NUM> provides a front end for dataflow services. Interface <NUM> may provide input and output for dataflow services. Interface <NUM> receives incoming dataflow plans (e.g., dataflow plan(s) <NUM>). Interface <NUM> may provide information and/or other feedback for display in a graphical user interface (GUI) provided by an operating system or application <NUM> executed by computing device(s) <NUM>.

Metadata generator <NUM> generates metadata for dataflows, for example, in preparation for weight generator <NUM>. Metadata may identify any aspect of importance to weight generator <NUM>. Metadata generator <NUM> may generate metadata, for example, at publish time (e.g., when a dataflow is published, before compiling and execution). Metadata may comprise, for example, a total number and type of each source, transformation and sink, type(s) of data, size(s) of data being consumed, processed and written, a number of different write operations (e.g., insert, update, delete, etc.) and so on. Metadata may be associated with a dataflow plan generally, by node, in a separate file, by association expected by weight generator <NUM>, and/or in other manners.

Weight generator <NUM> generates weights for each node (e.g., operator, transform, or step) in a dataflow plan. Weight generator <NUM> may generate weights, for example, based on metadata and/or other indicators of a dataflow plan. Weights generated for an unweighted dataflow may indicate, for example, varying levels of consumption of one or more resources. Weight values may be normalized, for example, on a scale of <NUM> to <NUM>. Weighting dataflow plans to determine resource configurations (e.g., in advance of execution) may be applied, for example, to any type of data processing pipeline.

Weight generator <NUM> may also provide weight adjustment or may add separate weights based on one or more user-specified preferences. In an example, application <NUM> may be configured to provide feedback, request and/or receive user preferences. Feedback may indicate (e.g., for a dataflow plan) estimated resources, run times and costs to let users choose between multiple IR configurations based on time sensitivity, cost sensitivity, or other preferences. For example, a GUI may display a dial or input box where a user may indicate a preference that can be fed into their model to determine whether to err on side of improved performance with more cost or more time and/or fewer resources with less cost.

Weight generator <NUM> may, for example, utilize a machine learning (ML) model to generate weights. A weight generating ML model may be trained based on historical dataflow plan metadata or other dataflow plan indicators for historical dataflow plans with known execution environments and known performance. Historical metadata may be used to create dataset features for training, which may be divided into training and validation sets (e.g., for cross validation). A trained and validated model may be used to infer weights (e.g., weight vectors) for dataflow plan(s) <NUM> received from user(s) <NUM>.

Weight generator <NUM> may featurize a selective dataflow (e.g., ETL) dataset in order to generate a weighted dataflow from an unweighted dataflow. Feature sets generated for training ML models and for received dataflow plan(s) <NUM> may be based on types of dataflows. An unlimited number of dataflows may have millions of different settings, not all of which may be pertinent or factored in to determining IRs. Feature tuning and/or filtering (e.g., as part of an ML model) may pare down features relevant to dataflow execution. For example, one or more features may indicate (e.g., to an ML model trained on historical dataflow features) that a majority of data from data sources may not be used in a dataflow plan. The number of nodes may vary in each dataflow. Feature extraction may extract features from wide variety of dataflow plans. Feature types and values may vary with plans.

Dataflow plan(s) <NUM> may be broken down into features in a feature set (e.g., following metadata association). In an example, labels for types of features may be, for example, F1 = type of transformation, F2 = number of joins, F3 = type of write operations (e.g., insert, update, delete), F4 = type of sources, F5 = type of sink, and so on. Feature entries may be values V1, V2, V3, V4, V5, etc. that represent or list a quantity/count of a type of feature, e.g., F2 is number of (e.g., <NUM>) joins.

After a feature set is generated for a dataflow plan, a classification model (e.g., supervised logic/algorithm, decision tree, neural network, ML model) that has been preconfigured (e.g., logically constructed or trained) may be used to classify the feature set into a weighted diagram. A model may be trained with similar feature sets extracted from a similar set of dataflows. Training sets may be manually created (e.g., tagged) with metadata, feature sets may be extracted, weight sets may be assigned for feature sets, and execution environments and performance results may be known.

In an example, a training set to train an ML model may comprise dataflow plan features paired with weight vector values associated with execution environments known to have performed well. Table <NUM> shows a simplified example of a model training set comprising multiple pairs of dataflow plan (DFP) feature sets and weight vectors. Feature values (e.g., based on plan metadata) may be paired with weight vector values to train a Weight Generator ML model to infer weight vectors for dataflow plan(s) <NUM>.

Multiple weight generator models may be present. In an example, a weight generator model may be selected based on metadata. Some features may be selected from a larger set of available features for a dataflow plan, for example, based on the importance of the feature(s) to a selected model. Feature selection may, for example, be based on term frequency-inverse document frequency (TF-IDF). A feature set may comprise zero or more TF-IDF entries.

A common feature may not indicate anything unique about a dataflow. In an example, a feature (Feature <NUM>) that occurs <NUM>,<NUM> times in a first dataflow and <NUM>,<NUM> or <NUM>,<NUM> times in other dataflows is not unique to the dataflow. Common features may have unique links to other features, which may be identified as features. Feature extraction may determine links between parts of dataflow, for example, to determine where features emanate from (e.g., feature1 is associated with a first source, feature2 is associated with a transformation, feature3 is associated with a join). A weight generator model may generate an array of weights (e.g., in vector form (<NUM>, <NUM>, <NUM>, <NUM>), for example, based on selected (e.g., unique) features. A (e.g., each) feature may correspond to one transformation in the dataflow. A unique feature may be selected for each transformation. Each block in the dataflow may be assigned a weight, resulting in a weighted graph created from an unweighted graph input by a user. In an example, the number of nodes in a dataflow may be the size (e.g., length) of a weight vector. In other example, weights may not be assigned per node.

TF-IDF may indicate what (if anything) is unique in a dataflow, for example, in addition to complex features of a dataflow, such as a number of joins. Not all dataflows with the same number of joins may be executed with the same IR, for example, based on unique features and/or other features detected by TF-IDF. In an example, dataflows with the same number of joins without a window may map to IR1 while those with a window may map to IR2.

Max flow detector (MFD) <NUM> is configured to determine a maximum flow (e.g., in terms of an amount of data in a network, in memory and or computation) for a received dataflow plan. For example, max flow detector <NUM> may apply a max-flow min-cut theorem to a weighted dataflow to determine the maximum flow. Furthermore, max flow detector <NUM> may be used to detect bottlenecks in a flow diagram. Max flow identification(s) may be utilized, for example, to avoid bottlenecks, e.g., by planning an execution environment in advance of execution that will avoid bottlenecks.

The highest weight(s) may not indicate the max flow in an MFD implementation. For example, there may be little effect from a maximum weight at the end of a dataflow process (e.g., writing to a database sink). Identifying bottlenecks in a process may be more important to identify max flow in some implementations.

Runtime mapper <NUM> may use a maximum flow determination to select (e.g., map to) an execution environment (e.g., an integration runtime (IR)), for example, from a predefined list or dictionary of execution environments. Each of multiple execution environments may be associated with or mapped to a range of max-flow values. Mapping to execution environments may be based on fine and/or coarse ranges. Table <NUM> shows an example mapping of max flow (MF) values to resource configurations (RCs) of resources.

Table <NUM> shows an example mapping of MF range <NUM> to <NUM> to a first resource configuration, a mapping of MF range <NUM> to <NUM> to a second resource configuration, a mapping of MF range <NUM> to <NUM> to a third resource configuration, and so on. Of course, this is one of and endless number of examples mapping weight vectors and/or max flow values to resource configurations.

Runtime mapper <NUM> may select an execution environment (e.g., IR (integration runtime) or RC (resource configuration)) before runtime (e.g., at compile time) for various reasons, such as to avoid runtime bottlenecks, delays, failures, and/or reallocations that occur when users select inadequate resources. The determination of an IR environment at compile time reduces the need to run time dependencies on external systems, thereby increasing the efficiencies and reliability of ETL pipelines.

Runtime mapper <NUM> may identify or select a best fit execution environment for a dataflow plan based on similarities between dataflow plan features, weight vectors and/or max flow detection relative to and feature sets, weight vectors and max flow detections that an ML model was trained on.

Selection of an execution environment is one of many types of classifications using a weighted data pipeline in combination with a max-flow detector. A maximum flow may be utilized to select a combination of computing resources to process a weighted pipeline. In an example, a combination of computing resources (e.g., number of servers, processor cores, memory at specific locations), storage resources (storage unit capacities and locations), and network resources (e.g., network interfaces and locations) may be selected based on a maximum flow determination for a weighted pipeline. For example, a best fit execution environment may be <NUM> processor core workers for <NUM> hours rather than <NUM> core workers at <NUM> hours costing four times more.

Runtime mapper <NUM> may be implemented, for example, with a predefined mapping dictionary, an ML model and/or other decision logic based on max flow, weight vectors and/or other information about a received dataflow plan.

Execution environments, integration runtimes (IRs) and resource configurations (RCs) may be referenced interchangeably. IRs may be configured and/or adjusted, for example, to be one or more of the following: compute optimized, memory optimized, storage optimized, and/or network optimized.

An execution environment (e.g., IR or RC) may indicate, for example, a number and type of resources for computing, memory, network, and storage resources. An execution environment may comprise, for example, five machines (computing devices), of which three machines with 32GB RAM, 512GB SSD, 1GB/s Internet throughput, and two machines with 16GB RAM, <NUM> GB SSD and <NUM> MB/s Internet throughput.

In an example, a diagramed flow may need more network throughput when data source(s) and processing are far away from each other. For example, a dataflow running in Western Europe that needs data stored in Eastern Europe or North America may need more network resources and/or faster network resources. In examples, a compute and/or memory optimized IR (e.g., memory optimized IR with <NUM> cores) may be selected for low data (e.g., a few gigabytes (GBs)) and heavy computing (e.g., many cross joins).

IR resources may be requested at compile time and constructed just in time (JIT) from datacenter (DC) resources. In an example, a data analytics platform (e.g., Microsoft ® Azure Databricks) may construct and configure resources (e.g., computing, VM, network, and storage resources) from a resource manager (e.g., Microsoft ® Azure Resource Manager). In an example (e.g., for a memory optimized IR with <NUM> cores), a data analytics platform executing on one or more dataflow server(s) <NUM> may fetch all the resources (e.g., <NUM> VMs with <NUM> GB memory and <NUM> GB throughput network) for the IR through a resource manager (e.g., resource manager(s) <NUM>).

Runtime mapper <NUM> may be limited (e.g., in IR selection), for example, by user (e.g., customer) related limitations. In an example, a service level agreement (SLA) may specify a minimum or a maximum number of workers per second. Auto-selection may increase or decrease one or more resources and/or select different IRs based on extraneous agreements, for example. Prevailing conditions or resource availability may also impact IR selections by runtime mapper <NUM>.

In an example, max flow detector <NUM> and runtime mapper <NUM> may be implemented as a (e.g., second) ML model, which may accept as input, for example, a weight vector and/or other configured input to generate as an output a classification in support of IR selection for received dataflow plan(s) <NUM>.

Additional information may be utilized (e.g., in training data sets and received dataflow plans), for example, by metadata generator <NUM>, weight generator <NUM> and/or IR selection by runtime mapper <NUM>. In an example, weights may be selected, at least in part, based on day, time of day, prevailing resource load, cost, etc..

<FIG> show examples of metadata and weights being generated, assigned and/or associated with nodes in a graph submitted as a dataflow plan. In other example implementations, metadata and/or weights may be assigned differently (e.g., based on edges, to a dataplan generally or as a whole, separately, such as in vectors). <FIG> are, therefore, one of many different example implementations.

For instance, <FIG> shows block diagram showing an example unweighted dataflow plan <NUM>, according to an example embodiment. Unweighted dataflow plan <NUM> may have been provided by user(s) <NUM> and received at interface <NUM>, for example. Unweighted dataflow plan <NUM> may comprise a graph with nodes and connecting edges between nodes. For example, in <FIG>, unweighted dataflow plan <NUM> includes first through fifth nodes N1-N5 that represent first through fifth data operations <NUM>-<NUM>. First and second operations O1 and O2 may represent data extraction operations from two data sources. Fifth operation O5 may represent a data writing operation of resulting data to a data sink (e.g., destination). In an example, a data source may comprise any type of storage, such as, a structured query language (SQL) database (DB) or data warehouse (DW), blob storage, cloud storage, etc..

<FIG> is a block diagram of an unweighted data flow plan <NUM> with associated dataflow metadata, according to an example embodiment. In particular, unweighted dataflow plan <NUM> is unweighted dataflow plan <NUM> of <FIG> with associated metadata M1-M5. In an example, metadata generator <NUM> may receive unweighted dataflow plan (e.g., input graph) <NUM> of <FIG> via interface <NUM>. Metadata generator <NUM> may generate and attach or associate metadata with input unweighted dataflow plan <NUM> based on the workflow specified by the input graph. Metadata generator <NUM> may associate metadata with nodes and/or with operations. For example, metadata may associate metadata M1-M5 with first through fifth nodes N1-N5 (first through fifth operations O1-O5). Metadata generator <NUM> may be configured to generate metadata relevant to weighting operations by weight generator <NUM>. For example, metadata may be configured based, at least in part, on features that a weighting ML model was trained on.

Metadata generator <NUM> may, for example, determine the number of sources, each type of source, the type and number of operations (e.g., the number of join operations or joins). Joins may consume more resources and time, which may lead to a larger workload to produce a node. Metadata may distinguish types of joins, which may have different workloads. For example, a join may be a cross join, where every possible combination on every row from each source may be joined. Metadata generator <NUM> may be configured to provide metadata M1-M5 based on what weight generator <NUM> (e.g., one or more ML models) may expect (e.g., to extract useful features).

<FIG> is a block diagram showing an example of a weighted dataflow plan <NUM>, according to an example embodiment. <FIG> shows weighted dataflow plan <NUM> with weights W1-W5 generated by weight generator <NUM>. For instance, weight generator <NUM> may receive unweighted dataflow plan <NUM> of <FIG> with associated metadata, for example, after metadata generator <NUM> associates metadata with unweighted dataflow plan <NUM> of <FIG>. Weight generator <NUM> may be trained to process metadata, for example, based on historical metadata for historical dataflow plans with historical execution environments and historical performance. Weight generator <NUM> may generate weights, for example, based on a decision tree, a neural network, a trained ML model or other logic implementation.

Weights may be associated with (e.g., assigned to) nodes (e.g., first through fifth nodes N1-N5). Weights may indicate how "heavy" (e.g., resource-consuming) operation(s) may be in order to generate a node, for example, in terms of operational resources, such as one or more of compute resources, network bandwidth, storage resources, etc..

Join operations may consume more resources and time, which may lead to higher weights being associated with join operations (e.g., to indicate it will take, relatively, more time and resources to produce a node relative to other nodes).

Weights may be assigned based on the number of data sources, types of data sources, operations (e.g., join, cross-join), total number of transformations, size of each data item in operation (e.g., cross-join on <NUM> MB or <NUM> GB of data).

In an example, weight generator <NUM> may assign the following weights to nodes N1-N5 of unweighted dataflow plan <NUM>: W1 = <NUM>, W2 = <NUM>, W3 = <NUM>, W4 = <NUM> and W5 = <NUM>. A higher weight associated with third node N3 may indicate a chokepoint that may need more resources than other nodes to avoid delays or failures during execution.

Runtime mapper <NUM> may receive an indication of weighted dataflow plan <NUM>, such as a weight vector. A max flow detector (e.g., max flow detector <NUM> of <FIG>) may use a max flow min cut theorem to find the max flow and the min cut (e.g., bottleneck or chokepoint). In an example, a weight of <NUM> may indicate a bottleneck, which may be used, at least in part, to determine the number and/or size of resources and/or execution environment needed for input dataflow plan <NUM>.

Runtime mapper <NUM> may utilize a predefined dictionary or mapping table that relates max flow values or ranges to execution environments (e.g., integration runtimes or IRs). Max flow values may be normalized, for example, between <NUM> and <NUM>. In an example, a first IR may be selected for a max flow range of <NUM> to <NUM>, a second IR may be selected for a max flow range of <NUM> to <NUM> and a third IR may be selected for a max flow range of <NUM> to <NUM>.

Implementations are not limited to the example shown in <FIG>. Any number of computing devices and/or servers (including but not limited to machines and/or virtual machines) may be coupled in any manner via any type of computing environment. For example, one or more of computing device, server or storage components may be co-located, located remote from each other, combined or integrated on or distributed across one or more real or virtual machines. Example system <NUM> or components therein may operate, for example, according to example methods presented in <FIG>.

Embodiments may also be implemented in processes or methods. For example, <FIG> is a flowchart <NUM> showing a method for automated runtime configuration for dataflows. A dataflow plan may include user preferences (e.g., pre-weights), which may influence metadata and weight generation and/or IR selection. Preferences may be specified initially or upon feedback and/or input request. Embodiments disclosed herein and other embodiments may operate in accordance with example flowchart <NUM>. Flowchart <NUM> comprises steps <NUM>-<NUM>. However, other examples may operate according to other methods. Other structural and operational embodiments will be apparent to persons skilled in the relevant art(s) based on the foregoing discussion of embodiments. No order of steps is required unless expressly indicated or inherently required. There is no requirement that a method embodiment implement all of the steps illustrated in <FIG> is simply one of many possible embodiments. Embodiments may implement fewer, more or different steps. Flowchart <NUM> is described as follows.

In step <NUM>, a dataflow plan (e.g., comprising data operations in a dataflow pipeline) is received. For example, as shown in <FIG> and <FIG>, dataflow server(s) <NUM> (e.g., interface <NUM>) may receive dataflow plan(s) <NUM> via computing device(s) <NUM> and network(s) <NUM>.

In step <NUM>, metadata (e.g., comprising information about each of the data operations) are generated for the received dataflow plan. For example, as shown in <FIG> and <FIG>, metadata generator <NUM> generates metadata for dataflow plan(s) <NUM> received by interface <NUM>.

In step <NUM>, a weighted dataflow plan is created (e.g., based on the received dataflow plan and metadata) by determining weights for the data operations based on the metadata. For example, as shown in <FIG> and <FIG>, weight generator <NUM> generates a weighted dataflow plan <NUM> based on metadata annotated dataflow plan <NUM>.

In step <NUM>, a maximum dataflow is determined for the weighted dataflow plan. For example, as shown in <FIG>, max flow detector <NUM> generates a max flow for weighted dataflow plan <NUM>.

In step <NUM>, at least one of a runtime environment or one or more runtime resources are selected to execute the received dataflow plan based on the weighted dataflow plan and/or the maximum dataflow. For example, as shown in <FIG>, runtime mapper <NUM> may select an execution environment (e.g., an IR or RC) for dataflow plan(s) <NUM> received by interface <NUM> based on weighted dataflow plan <NUM> and/or max flow determined by max flow detector <NUM>.

<FIG> is a flowchart <NUM> showing an example method for automated runtime configuration for dataflows, according to an example embodiment. <FIG> shows example detail for the example method shown in <FIG>, for example, when utilizing ML models with feature sets to determine weights and runtime configurations for dataflow plans. Embodiments disclosed herein and other embodiments may operate in accordance with example flowchart <NUM>. Flowchart <NUM> comprises steps <NUM>-<NUM>. However, other embodiments may operate according to other methods. Other structural and operational embodiments will be apparent to persons skilled in the relevant art(s) based on the foregoing discussion of embodiments. No order of steps is required unless expressly indicated or inherently required. There is no requirement that a method embodiment implement all of the steps illustrated in <FIG> is simply one of many possible embodiments. Embodiments may implement fewer, more or different steps. Flowchart <NUM> is described as follows.

In step <NUM>, a feature set (e.g., a plurality of features) may be generated from the received dataflow plan. For example, as shown in <FIG>, weight generator <NUM>, implemented as a first trained ML model, may generate a feature set from dataflow plan(s) <NUM> received by interface <NUM> (e.g., and annotated with metadata by metadata generator <NUM>).

In step <NUM>, a first machine learning (ML) model may generate the weighted dataflow plan as a weighted feature set based on the feature set. For example, as shown in <FIG>, weight generator <NUM>, implemented as a first trained ML model, may generate a weighted dataflow plan as a weighted feature set based on the feature set extracted from dataflow plan(s) <NUM> received by interface <NUM> (e.g., and annotated with metadata by metadata generator <NUM>).

In step <NUM>, the weighted feature set may be provided to a second ML model. For example, as shown in <FIG>, the weighted feature set generated by weight generator <NUM> may be provided to a second ML model comprising runtime mapper <NUM> (e.g., merged with max flow detector <NUM>).

In step <NUM>, the second ML model may determine the runtime environment or the runtime resources based on the weighted feature set. For example, as shown in <FIG>, runtime mapper <NUM> implemented as a second ML model may receive as input a weighted feature set (e.g., as a vector) and output a classification of the weighted feature set as one or more IRs (e.g., with a confidence level associated with each IR classification).

As noted herein, the embodiments described, along with any modules, components and/or subcomponents thereof (e.g., resource manager(s) <NUM>, resources <NUM>, dataflow server(s) <NUM>, interface <NUM>, metadata generator <NUM>, weight generator <NUM>, max flow detector <NUM>, runtime mapper <NUM>, computing device(s) <NUM>, application <NUM>, dataflow plan(s) <NUM>, data source(s) <NUM>, dataflow transformation <NUM>, and/or data sink(s) <NUM>), as well as the flowcharts/flow diagrams described herein (e.g., , flowchart <NUM> and/or flowchart <NUM>), including portions thereof, and/or other embodiments, may be implemented in hardware, or hardware with any combination of software and/or firmware, including being implemented as computer program code configured to be executed in one or more processors and stored in a computer readable storage medium, or being implemented as hardware logic/electrical circuitry, such as being implemented together in a system-on-chip (SoC), a field programmable gate array (FPGA), and/or an application specific integrated circuit (ASIC). A SoC may include an integrated circuit chip that includes one or more of a processor (e.g., a microcontroller, microprocessor, digital signal processor (DSP), etc.), memory, one or more communication interfaces, and/or further circuits and/or embedded firmware to perform its functions.

<FIG> shows an exemplary implementation of a computing device <NUM> in which example embodiments may be implemented. For instance, one or more of computing device(s) <NUM>, resource manager(s) <NUM>, resources <NUM>, data source(s) <NUM>, dataflow transformations <NUM>, data sink(s) <NUM>, and/or dataflow server(s) <NUM> may include features of computing device <NUM>. Consistent with all other descriptions provided herein, the description of computing device <NUM> is a non-limiting example for purposes of illustration. Example embodiments may be implemented in other types of computer systems, as would be known to persons skilled in the relevant art(s).

A number of program modules may be stored on the hard disk, magnetic disk, optical disk, ROM, or RAM. These programs include operating system <NUM>, one or more application programs <NUM>, other programs <NUM>, and program data <NUM>. Application programs <NUM> or other programs <NUM> may include, for example, computer program logic (e.g., computer program code or instructions) for implementing resource manager(s) <NUM>, resources <NUM>, dataflow server(s) <NUM>, interface <NUM>, metadata generator <NUM>, weight generator <NUM>, max flow detector <NUM>, runtime mapper <NUM>, computing device(s) <NUM>, application <NUM>, dataflow plan(s) <NUM>, data source(s) <NUM>, dataflow transformation <NUM> and data sink(s) <NUM>, flowchart <NUM>, and/or flowchart <NUM> (including any suitable step of flowcharts <NUM> or <NUM>) and/or further example embodiments described herein.

Example embodiments are also directed to such communication media that are separate and non-overlapping with embodiments directed to computer-readable storage media.

Such computer programs, when executed or loaded by an application, enable computing device <NUM> to implement features of example embodiments described herein.

Example embodiments are also directed to computer program products comprising computer code or instructions stored on any computer-readable medium.

Methods, systems and computer program products are provided for automated runtime configuration for dataflows, for example, to automatically select or adapt a runtime environment or resources to a dataflow plan (e.g., expressing data operations) prior to execution (e.g., to eliminate guesswork, avoid runtime failures and scaling). Data operations may comprise, for example, extracting input data from source(s), transforming input data into transformed data, and loading input data and/or transformed data into sink(s). Metadata may be generated for dataflows to indicate numbers and types of sources, sinks and operations, and the amount of data being consumed, processed and written. Weighted dataflow plans may be created from unweighted (e.g., codeless graphical) dataflow plans based on metadata. Weights that indicate operation complexity or resource consumption may be generated (e.g., by a trained model) and associated with data operations (e.g., features). A maximum flow may be determined from a weighted dataflow (e.g., using max flow min cut to identify bottlenecks). A runtime environment or resources to execute a dataflow plan may be selected based on the maximum flow. Runtime selections may be influenced by preferences (e.g., weights for cost or time sensitivity).

In an example, a system for automatic runtime environment or resource selection for a dataflow may comprise, for example, a metadata generator, a weight generator, a max flow detector and a mapper. A metadata generator may be configured, for example, to receive a dataflow plan comprising data operations in a dataflow pipeline; and generate metadata for the received dataflow plan comprising information about each of the data operations. A weight generator may be configured, for example, to create a weighted dataflow plan from the received dataflow plan by associating a weight with each of the data operations based on the metadata. A max flow detector may be configured, for example, to determine a maximum dataflow for the weighted dataflow plan. A mapper may be configured, for example, to select at least one of a runtime environment or one or more runtime resources to execute the received dataflow plan based at least on the maximum dataflow.

In an example, the weight generator is configured to generate a feature set comprising a plurality of features from the received dataflow plan. The weight generator is configured to create the weighted dataflow plan by generating a feature weight set for the feature set based on the metadata.

In an example, the max flow detector may be configured, for example, to determine the maximum dataflow based on application of a max-flow min-cut theorem to the weighted dataflow plan.

In an example, a method of automatic runtime environment or resource selection for a dataflow may comprise, for example, receiving a dataflow plan comprising data operations in a dataflow pipeline; creating a weighted dataflow plan from the received dataflow plan by associating a weight with each of the data operations; and selecting at least one of a runtime environment or one or more runtime resources to execute the received dataflow plan based at least on the weighted dataflow plan. This method may automatically and proactively, adapt or select a runtime environment or resources for a dataflow plan prior to execution, which may reduce or eliminate runtime guesswork, failures and/or scaling for dataflow execution of a dataflow plan.

In an example, a weight associated with a data operation may be selected from a weight range corresponding to an execution complexity range for each of the data operations in terms of consumption of at least one of computing, memory, storage and network resources.

In an example, a weight associated with a data operation may comprise a combined weight or a weight vector indicating at least two of computing, memory, storage and network resources.

In an example, the method may further comprise, for example, generating metadata for the received dataflow plan comprising information about each of the data operations. The weight associated with each of the data operations may be based on the metadata.

In an example, the method may further comprise, for example, determining a maximum dataflow for the weighted dataflow plan. The selection of the runtime environment or the runtime resources may be based on the maximum dataflow. In an example, a maximum dataflow may be defined, for example, in units of data sizes or a number of rows processed (e.g., based on type of data source(s)).

In an example, determining the maximum dataflow may comprise applying a max-flow min-cut theorem to the weighted dataflow plan.

In an example, the method may further comprise, for example, selecting the runtime environment or the runtime resources based on an indication of a preference related to at least one of execution time and execution cost to execute the received dataflow plan.

In an example, the method may further comprise, for example, providing feedback to a user interface indicating a plurality of runtime environments or a plurality of runtime resource alternatives and their costs in terms of at least one of time and price; and prompting user input to select from among the plurality of runtime environments or the plurality of runtime resource alternatives to execute the received dataflow plan.

In an example, the selection of the runtime environment or the runtime resources may comprise, for example, selecting between at least two of: compute optimized, memory optimized, network optimized, and storage optimized runtime environments or runtime resources.

In an example, the received dataflow plan may comprise a graph with nodes and edges provided by a user.

In an example, the method may further comprise, for example, restricting the selection of the runtime environment or the runtime resources based on one or more caps or limits specified in a user agreement.

In an example, the method comprises, generating a feature set comprising a plurality of features from the received dataflow plan; and creating the weighted dataflow plan by providing the feature set as input a first machine learning (ML) model that generates a feature weight set.

In an example, the method may further comprise, for example, providing the feature weight set to a second ML model; and selecting the runtime environment or the runtime resources based on an output of the second ML model.

In an example, at least one feature in the plurality of features may be generated based on application of term frequency-inverse document frequency (TF-IDF) to the received dataflow plan.

In an example, a computer-readable storage medium may have program instructions recorded thereon that, when executed by a processing circuit, perform a method comprising: receiving a dataflow plan comprising data operations in a dataflow pipeline; generating metadata for the received dataflow plan comprising information about each of the data operations; creating a weighted dataflow plan from the received dataflow plan by associating a weight with each of the data operations based on the metadata; and selecting at least one of a runtime environment or one or more runtime resources to execute the received dataflow plan based at least on the weighted dataflow plan.

In an example, the method may further comprise, for example, determining a maximum dataflow for the weighted dataflow plan. The selection of the runtime environment or the runtime resources may be based on the maximum dataflow.

In an example, the weighted dataflow plan may be created based on the output of a first machine learning (ML) model. The runtime environment or the runtime resources may be selected based on the output of a second ML model.

Claim 1:
A system, comprising:
a metadata generator (<NUM>) configured to:
receive (<NUM>) a dataflow plan (<NUM>) comprising data operations in a dataflow pipeline, wherein the dataflow plan indicates one or more data sources (<NUM>), one or more dataflow transformations (<NUM>) and one or more data sinks (<NUM>), wherein the data operations comprise at least one of extracting input data from the one or more data sources, transforming the input data into transformed data, and loading the input data or the transformed data into the one or more data sinks; and
generate (<NUM>) metadata for the received dataflow plan comprising information about each of the data operations, the one or more data sources and the one or more data sinks;
a weight generator (<NUM>) configured to:
generate a feature set comprising a plurality of features from the received dataflow plan, wherein each feature of the plurality of features corresponds to one of a type of a dataflow transformation, a number of join operations, a type of a write operation, a type of a data source, and a type of a data sink; and
create (<NUM>) a weighted dataflow plan from the received dataflow plan by generating a feature weight set for the feature set based on the metadata using a trained machine learning, ML, model and associating a weight with each of the data operations based on the generated feature weight set;
a max flow detector (<NUM>) configured to:
determine (<NUM>) a maximum dataflow for the weighted dataflow plan; and
a mapper (<NUM>) configured to:
select (<NUM>) at least one of a runtime environment or one or more runtime resources to execute the received dataflow plan based at least on the maximum dataflow.