Connects and starts if needed the PigServer. @return Reference to the Cluster in ThreadLocal. @throws ExecException If the PigServer can't be started.
()
| 129 | * @throws ExecException If the PigServer can't be started. |
| 130 | */ |
| 131 | public static Cluster getCluster() throws ExecException { |
| 132 | try { |
| 133 | if (cluster.get() == null) { |
| 134 | ExecType execType = ExecType.LOCAL; |
| 135 | if (System.getProperties().containsKey(EXEC_CLUSTER)) { |
| 136 | if (System.getProperties().getProperty(EXEC_CLUSTER).equalsIgnoreCase("mr")) { |
| 137 | LOG.info("Using mr cluster mode"); |
| 138 | execType = ExecType.MAPREDUCE; |
| 139 | } else if (System.getProperties().getProperty(EXEC_CLUSTER).equalsIgnoreCase("tez")) { |
| 140 | LOG.info("Using tez cluster mode"); |
| 141 | execType = ExecTypeProvider.fromString("tez"); |
| 142 | } else if (System.getProperties().getProperty(EXEC_CLUSTER).equalsIgnoreCase("tez_local")) { |
| 143 | LOG.info("Using tez local mode"); |
| 144 | execType = ExecTypeProvider.fromString("tez_local"); |
| 145 | } else if (System.getProperties().getProperty(EXEC_CLUSTER).equalsIgnoreCase("spark")) { |
| 146 | LOG.info("Using spark cluster mode"); |
| 147 | execType = ExecTypeProvider.fromString("spark"); |
| 148 | } else if (System.getProperties().getProperty(EXEC_CLUSTER).equalsIgnoreCase("spark_local")) { |
| 149 | LOG.info("Using spark local cluster mode"); |
| 150 | execType = ExecTypeProvider.fromString("spark_local"); |
| 151 | } else { |
| 152 | LOG.info("Using default local mode"); |
| 153 | } |
| 154 | } else { |
| 155 | LOG.info("Using default local mode"); |
| 156 | } |
| 157 | pig.set(new PigServer(execType)); |
| 158 | cluster.set(new Cluster(pig.get().getPigContext())); |
| 159 | } |
| 160 | } catch (PigException e) { |
| 161 | throw new ExecException(e); |
| 162 | } |
| 163 | |
| 164 | return cluster.get(); |
| 165 | } |
| 166 | |
| 167 | /** |
| 168 | * Return the PigServer. |
no test coverage detected