yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-12-18 21:44:03 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #808 from camelliazhang/master

Browse Source 
fix a  bug in ETL
This commit is contained in:
na zhang 2017-10-19 14:07:28 -07:00 committed by GitHub
commit ba7082a787
1 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
wherehows-etl/src/main/resources/jython/ElasticSearchIndex.py
View File

@ -252,7 +252,6 @@ class ElasticSearchIndex():
url = self.elasticsearch_index_url + ':' + str( url = self.elasticsearch_index_url + ':' + str(
self.elasticsearch_port) + '/' + self.elasticsearch_index + '/dataset/_bulk' self.elasticsearch_port) + '/' + self.elasticsearch_index + '/dataset/_bulk'
params = [] params = []
while result: while result:
row = dict(zip(description, result)) row = dict(zip(description, result))
@ -276,12 +275,13 @@ class ElasticSearchIndex():
params.append('{ "index": { "_id": ' + str(row['id']) + ' }}') params.append('{ "index": { "_id": ' + str(row['id']) + ' }}')
params.append(json.dumps(dataset_detail)) params.append(json.dumps(dataset_detail))
if row_count % self.bulk_chunk_size == 0: if row_count % self.bulk_chunk_size == 0:
self.bulk_insert(params, url) self.bulk_insert(params, url)
self.logger.info('dataset ' + str(row_count)) self.logger.info('dataset ' + str(row_count))
self.wh_con.commit() self.wh_con.commit()
params = [] params = []
row_count += 1 row_count += 1
result = self.wh_cursor.fetchone() result = self.wh_cursor.fetchone()
self.logger.info('total dataset row count is: ' + str(row_count)) self.logger.info('total dataset row count is: ' + str(row_count))
@ -344,8 +344,8 @@ class ElasticSearchIndex():
self.bulk_insert(params, url) self.bulk_insert(params, url)
self.logger.info('metric ' + str(row_count)) self.logger.info('metric ' + str(row_count))
self.wh_con.commit() self.wh_con.commit()
params = [] params = []
row_count += 1 row_count += 1
result = self.wh_cursor.fetchone() result = self.wh_cursor.fetchone()
if len(params) > 0: if len(params) > 0: