Add test case illustrating slow planning

[mdelaurentis]

We have found an issue where certain conditions in a flow can cause the planning phase to take a very long time. It is related to flows that use a lot of CoGroups, but there may be other conditions that trigger it. The bulk of the time is spent in calls to methods on ElementGraphs and ElementGraph that find the shortest path between pairs of nodes. These calls are made by HadoopPlanner, LocalPlanner, and FlowPlanner. As we add more CoGroups, it looks like we make many more calls to KShortestPaths.getPaths, and each call takes much longer.

We have a flow that we are building dynamically for a client, based on user configuration. They have many tables (about 50, I think), and unfortunately the flow as it's currently configured results in a lot of joins between those tables. In some cases the same two tables are joined together several times using different key fields. We started the job almost 48 hours ago, and it is still in the planning phase. We can probably modify our code that translates user configuration into a Flow so that we reduce the number of joins, but it still seems like it is taking far longer than it should to build this flow.

This test case should illustrate the problem. I'm just creating n source taps, n sink taps, and n pipes. For each i > 1, pipe i is joined onto pipe i - 1. I timed the calls to connect() for various values of i. Those timings are listed below. Note that the flow we're actually trying to run is more complex than this one, as it isn't simply chaining together n tables. It seems to actually scale worse than this contrived example.

I have an idea for a change that might speed it up, but I'm not sure if it would be acceptable. If you can get away with just one of the shortest paths instead of all the shortest paths between a pair of nodes, you should be able to use JGraphT's FloydWarshallShortestPaths class. That uses a dynamic programming algorithm to find the shortest path between every pair of n nodes in O(n^3) time and stores them in a table. I have experimented with this locally and it seems that at least for the flow we're having trouble with it will speed things up dramatically. I was able to run the Floyd-Warshall algorihtm on the graph for this flow in less than a minute. I have been running an unpatched Cascading 2.2 job on the same flow, and it has been sitting in the planning phase for almost 48 hours now.

n = 1: 0.008 seconds
n = 2: 0.016 seconds
n = 3: 0.025 seconds
n = 4: 0.038 seconds
n = 5: 0.056 seconds
n = 6: 0.066 seconds
n = 7: 0.057 seconds
n = 8: 0.066 seconds
n = 9: 0.073 seconds
n = 10: 0.085 seconds
n = 11: 0.094 seconds
n = 12: 0.109 seconds
n = 13: 0.139 seconds
n = 14: 0.166 seconds
n = 15: 0.178 seconds
n = 16: 0.214 seconds
n = 17: 0.249 seconds
n = 18: 0.299 seconds
n = 19: 0.362 seconds
n = 20: 0.437 seconds
n = 21: 0.527 seconds
n = 22: 0.644 seconds
n = 23: 0.787 seconds
n = 24: 0.946 seconds
n = 25: 1.169 seconds
n = 26: 1.406 seconds
n = 27: 1.678 seconds
n = 28: 2.020 seconds
n = 29: 2.461 seconds
n = 30: 3.016 seconds
n = 31: 3.703 seconds
n = 32: 4.280 seconds
n = 33: 5.131 seconds
n = 34: 5.925 seconds
n = 35: 7.123 seconds
n = 36: 8.055 seconds
n = 37: 9.930 seconds
n = 38: 10.416 seconds
n = 39: 13.129 seconds
n = 40: 13.740 seconds
n = 41: 14.841 seconds
n = 42: 16.492 seconds
n = 43: 18.554 seconds
n = 44: 21.692 seconds
n = 45: 22.799 seconds
n = 46: 26.360 seconds
n = 47: 27.716 seconds
n = 48: 31.719 seconds
n = 49: 35.187 seconds
n = 50: 40.189 seconds
n = 51: 44.004 seconds
n = 52: 49.871 seconds
n = 53: 59.218 seconds
n = 54: 56.836 seconds
n = 55: 60.778 seconds
n = 56: 71.670 seconds
n = 57: 74.854 seconds
n = 58: 84.976 seconds
n = 59: 85.830 seconds
n = 60: 91.886 seconds
n = 61: 107.949 seconds
n = 62: 119.523 seconds
n = 63: 119.880 seconds
n = 64: 125.984 seconds
n = 65: 133.999 seconds
n = 66: 154.024 seconds
n = 67: 163.039 seconds
n = 68: 178.564 seconds
n = 69: 193.835 seconds
n = 70: 205.415 seconds
n = 71: 221.935 seconds
n = 72: 239.641 seconds
n = 73: 259.152 seconds
n = 74: 297.337 seconds
n = 75: 301.003 seconds
n = 76: 332.645 seconds
n = 77: 359.344 seconds
n = 78: 386.505 seconds
n = 79: 629.116 seconds

[mdelaurentis]

Oh, the test case I added has n set to 100, so running the tests will probably make it hang for hours. I'm not sure how you typically test these time-dependent kinds of things, but it seemed like this would be a reasonable way to show what's going on.

[dvryaboy]

@cwensel some of our users are reporting issues that look pretty similar to what this is purported to fix. Have you had a chance to look at the pull request?

[cwensel]

There is no fix here that I see. just a test. the suggestion above won't work, we need all paths.

I've removed the worst of it in the Cascading 3 progress i've made so far. we no longer need to traverse the paths except in narrow cases (and maybe we can remove those at some point).

but joining a 100 or a thousand taps will always be "slow" in the planning.

[dvryaboy]

Sorry, didn't look at the PR.

[cwensel]

so poking at the 3.0 planner i’m getting the following numbers (vs Cascading 2.5)

Cascading 2.5
n = 30: 8.645 seconds
n = 50: 35.908 seconds
n = 75: 314.075 seconds

Cascading 3.0
n = 25: elements: 123: 2.798 seconds
n = 30: elements: 148: 3.880 seconds
n = 50: elements: 248: 10.476 seconds
n = 75: elements: 373: 35.415 seconds
n = 100: elements: 498: 87.206 seconds
n = 200: elements: 998: 1327.832 seconds

still not perfect, and I think there are more opportunities for optimization to explore yet. but I am making progress. the number of elements is total tap+pipes added per n value.

I know I keep saying this, but I hope we can get 3.0 wips out this week for people to play with.

am going to close this. but if I get any more significant improvements, i’ll update this issue

[cwensel]

couldn’t help it. one more minor boost..

Cascading 3.0
n = 100: elements: 498: 83.207 seconds
n = 200: elements: 998: 1180.938 seconds