def METHOD_NAME(self, data_source: ElasticsearchDataSource) -> pd.DataFrame:
    data_source.body = nosql_apply_parameters_to_query(data_source.body, data_source.parameters)
    connection_params = []
    for host in self.hosts:
        parsed_url = urlparse(host.url)
        h = {'host': parsed_url.hostname}
        if parsed_url.path and parsed_url.path != '/':
            h['url_prefix'] = parsed_url.path
        if parsed_url.scheme == 'https':
            h['port'] = host.port or 443
            h['use_ssl'] = True
            h['scheme'] = parsed_url.scheme
        elif host.port:
            h['port'] = host.port
            h['scheme'] = parsed_url.scheme
        if host.username or host.password:
            h['http_auth'] = f'{host.username}:{host.password.get_secret_value()}'
        if host.headers:
            h['headers'] = host.headers
        connection_params.append(h)
    esclient = Elasticsearch(connection_params)
    response = getattr(esclient, data_source.search_method)(
        index=data_source.index, body=data_source.body
    )
    if data_source.search_method == SearchMethod.msearch:
        res = []
        # Body alternate index and query `[index, query, index, query...]`
        queries = data_source.body[1::2]
        for query, data in zip(queries, response['responses']):
            res += _read_response(data)
    else:
        res = _read_response(response)
    df = json_normalize(res)
    return df