PySpark으로 아래와 같이 코딩해서 데이터베이스에 올리고 있습니다.
q = """SELECT
NG.srv_id
,NG.item_id
,N.item_name
,N.ins_time
,NG.count
,NG.min_value
,NG.max_value
,NG.sum_value
,NG.avg_value
,N.host_id
,N.val_type
,N.host_name
,N.group_id
,N.group_name
,NG.start_time
,NG.end_time
FROM networkswitch_grpby NG
LEFT OUTER JOIN networkswitch N on NG.item_id=N.item_id AND NG.srv_id = N.srv_id AND NG.start_time = N.ins_time
ORDER BY NG.srv_id asc
"""
res = sqlContext.sql(q)
res.write.mode("append").format("jdbc")\
.option("url", db_url)\
.option("dbtable", table_name)\
.option("user", user)\
.option("password", password)\
.option("driver", driver)\
.save()
그런데 문제가 마지막에 res.write.mode("append") 하는 부분에서 append를 하면 이미 같은 데이터가 있음에도 불구하고 insert 되어 버리고 overwrite를 하면 당연하겠지만 데이터가 축적 되는 것이 아닌 덮어 써버리기 때문에 데이터 수집을 시간 단위로 해야하는 정보를 위해 사용할 수 없습니다. error나 디폴트 값으로 .mode를 동작하게 하면
데이터가 테이블에 아예 없음에도 불구하고 이미 데이터베이스가 exists한다고 아예 insert조차 되질 않네요.
1시간마다 데이터를 수집해서 insert시켜야 하는데 행여나 중복 된 데이터일 경우 중복을 감지하고 insert시키지 않고 새로운 데이터만 넣고싶은데 어떻게 수정하면 좋을까요?
참고로 데이터베이스는 MSSQL을 사용하고 srv_id, item_id, host_id, ins_time(수집시간) 네 개의 컬럼에 클러스터 프라이머리 키를 지정해두었습니다.