Creates a connection to an Impalad instance. Returns a tuple with the impala version string and the webserver address, otherwise raises an exception. If the client was already connected, closes the previous connection.
(self)
| 215 | self.reuse_http_connection = reuse_http_connection |
| 216 | |
| 217 | def connect(self): |
| 218 | """Creates a connection to an Impalad instance. Returns a tuple with the impala |
| 219 | version string and the webserver address, otherwise raises an exception. If the client |
| 220 | was already connected, closes the previous connection.""" |
| 221 | self.close_connection() |
| 222 | |
| 223 | try: |
| 224 | if self.use_http_base_transport: |
| 225 | self.transport = self._get_http_transport(self.client_connect_timeout_ms) |
| 226 | else: |
| 227 | self.transport = self._get_transport(self.client_connect_timeout_ms) |
| 228 | assert self.transport and self.transport.isOpen() |
| 229 | except TTransportException as e: |
| 230 | # Unwrap socket.error so we can handle it directly. |
| 231 | if isinstance(e.inner, socket.error): |
| 232 | raise e.inner |
| 233 | raise |
| 234 | if self.verbose: |
| 235 | msg = 'Opened TCP connection to %s:%s' % (self.impalad_host, self.impalad_port) |
| 236 | print(msg, file=sys.stderr) |
| 237 | protocol = TBinaryProtocol.TBinaryProtocolAccelerated(self.transport) |
| 238 | self.imp_service = self._get_thrift_client(protocol) |
| 239 | self.connected = True |
| 240 | try: |
| 241 | self._open_session() |
| 242 | return self._ping_impala_service() |
| 243 | except Exception: |
| 244 | # Ensure we are in a disconnected state if we failed above. |
| 245 | self.close_connection() |
| 246 | raise |
| 247 | |
| 248 | def _get_thrift_client(self, protocol): |
| 249 | """Instantiate a thrift client with the provided protocol.""" |
no test coverage detected