yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-09-26 17:34:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix #384: Check if the elastic search index exists and make sure we have document mappings (#386)

Browse Source
This commit is contained in:
Sriharsha Chintalapani 2021-09-02 20:28:26 -07:00 committed by GitHub
parent 26fc512501
commit 044b785d15
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 18 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
ingestion/src/metadata/ingestion/sink/elasticsearch.py
View File

@ -12,7 +12,7 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and # See the License for the specific language governing permissions and
# limitations under the License. # limitations under the License.
import json
import logging import logging
import time import time
from typing import Optional from typing import Optional
@ -82,11 +82,16 @@ class ElasticsearchSink(Sink):
Retrieve all indices that currently have {elasticsearch_alias} alias Retrieve all indices that currently have {elasticsearch_alias} alias
:return: list of elasticsearch indices :return: list of elasticsearch indices
""" """
try: if self.elasticsearch_client.indices.exists(index_name):
indices = self.elasticsearch_client.indices.get_alias(index_name).keys() mapping = self.elasticsearch_client.indices.get_mapping()
except NotFoundError: if not mapping[index_name]['mappings']:
logger.warn("Received index not found error from Elasticsearch. " logger.debug(f'There are no mappings for index {index_name}. Updating the mapping')
+ "The index doesn't exist for a newly created ES. It's OK on first run.") es_mapping_dict = json.loads(es_mapping)
es_mapping_update_dict = {'properties': es_mapping_dict['mappings']['properties']}
self.elasticsearch_client.indices.put_mapping(index=index_name, body=json.dumps(es_mapping_update_dict))
else:
logger.warning("Received index not found error from Elasticsearch. "
+ "The index doesn't exist for a newly created ES. It's OK on first run.")
# create new index with mapping # create new index with mapping
self.elasticsearch_client.indices.create(index=index_name, body=es_mapping) self.elasticsearch_client.indices.create(index=index_name, body=es_mapping)

9
ingestion/src/metadata/ingestion/source/metadata.py
View File

@ -40,10 +40,14 @@ class MetadataSourceStatus(SourceStatus):
failures: List[str] = field(default_factory=list) failures: List[str] = field(default_factory=list)
warnings: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list)
def scanned(self, table_name: str) -> None: def scanned_table(self, table_name: str) -> None:
self.success.append(table_name) self.success.append(table_name)
logger.info('Table Scanned: {}'.format(table_name)) logger.info('Table Scanned: {}'.format(table_name))
def scanned_topic(self, topic_name: str) -> None:
self.success.append(topic_name)
logger.info('Topic Scanned: {}'.format(topic_name))
def filtered(self, table_name: str, err: str, dataset_name: str = None, col_type: str = None) -> None: def filtered(self, table_name: str, err: str, dataset_name: str = None, col_type: str = None) -> None:
self.warnings.append(table_name) self.warnings.append(table_name)
logger.warning("Dropped Table {} due to {}".format(table_name, err)) logger.warning("Dropped Table {} due to {}".format(table_name, err))
@ -81,9 +85,10 @@ class MetadataSource(Source):
def next_record(self) -> Iterable[Record]: def next_record(self) -> Iterable[Record]:
for table in self.tables: for table in self.tables:
self.status.scanned(table.name.__root__) self.status.scanned_table(table.name.__root__)
yield table yield table
for topic in self.topics: for topic in self.topics:
self.status.scanned_topic(topic.name.__root__)
yield topic yield topic
def get_status(self) -> SourceStatus: def get_status(self) -> SourceStatus: