pyspark sqlContext로 데이터 삽입할 때 중복 되는 부분 없이 올리는 방법이 어떻게 되나요?

Question

pyspark sqlContext로 데이터 삽입할 때 중복 되는 부분 없이 올리는 방법이 어떻게 되나요?

조회수 465회

python

sql

insert

update

0

싫어요

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(수집시간) 네 개의 컬럼에 클러스터 프라이머리 키를 지정해두었습니다.

ignore 로 해보세요. 정영훈 2020.7.2 09:18
ignore로 하니까 아예 삽입이 안 되고 그냥 끝내버리네요... error로 적용해서 오류 원인을 보니까 이미 테이블이 존재한다고 오류를 보내네요... 테이블이 존재할 때는 append해서 데이터가 중복 될 때는 ignore이나 error를 반환해야 되는데 혹시 방법을 아시나요? 김재민 2020.7.2 09:38
애초에 쿼리 작성을 존재하지 않는 것으로 작성해서 추출하면 될 것 같은데요? 정영훈 2020.7.2 11:36
제공되는 mode 가 4가지 밖에 없어서 그외의 동작은 구현해줄수 밖에 없습니다. 정영훈 2020.7.2 11:38