173만명의 커뮤니티!! 함께 토론해봐요.
미해결
카프카 완벽 가이드 - 커넥트(Connect) 편
안녕하세요, postgresql의 logical replication에서는 publication이라고하는 모듈을 통해 복제 대상 테이블을 정해놓는데요 이렇게 되면 WAL에는 대상 테이블의 변경 내용만 담기는 건가요? 그리고 mysql에는 publication과 같은 모듈이 없는 것 같은데 bin 로그에는 그럼 모든 테이블의 트랜잭션 로그가 담기는 건지 궁금합니다.
미해결
카프카 완벽 가이드 - 커넥트(Connect) 편
안녕하세요, 디비지움 소스 테이블 스키마 변경에 따른 자동 스키마 반영에 대한 질문이 있습니다. 소스 테이블에 변경이 있을 때 예를 들어 varchar 타입 컬럼에 default를 적용하는 변경을 수행하는 경우 sink 쪽에서 오류가 발생하는 것으로 알고 있습니다. 그런데 만약 디비지움에 스키마 레지스트리가 적용되어 있고, 스키마 호환성을 체크한다고 한다면 디비지움의 소스 테이블 변경 규칙은 무시될수도 있으며, 스키마 레지스트리 호환성에 따라 동작하게 되는 것인가요?
미해결
카프카 완벽 가이드 - 커넥트(Connect) 편
안녕하세요, 스키마 레지스트리의 compatibility가 BACKWARD로 설정되어 있는 상태에서 nullable int 타입 컬럼을 source 쪽에서 추가하셨는데, BACKWARD 호환성의 경우는 읽기 쪽부터 변경스키마를 반영해야 하는 것이 아닌가요? nullable이므로 default가 있는 컬럼의 추가라서 BACKWARD를 만족한다는게 왜 소스쪽에서 추가한 내용 바탕으로 설명이 되는 것인지 이해가 되지 않습니다.
미해결
[아파치 카프카 애플리케이션 프로그래밍] 개념부터 컨슈머, 프로듀서, 커넥트, 스트림즈까지!
안녕하세요. 카프카 클러스터 내에 여러 개의 브로커를 실행한 다음 토픽의 파티션을 복제하고 리더 파티션과 팔로워 파티션이 있는 구조를 구성해보고 싶습니다. 위와 같은 구조를 직접 구현해보고 싶은데 혹시 참고할 만한 게 있을까요? 감사합니다.
미해결
mongoDB 기초부터 실무까지(feat. Node.js)
안녕하세요 강사님 !! ㅎㅎ 강의 재밌게 듣고 있습니다. async await 방식이 코틀린의 코루틴과 굉장히 유사하다고 생각이 되는데 그럼 다음과 같이 Promise.all 대신 await를 늦추는 방식으로 풀어써도 똑같이 동작하나요?? async createBlog() => { await Promise.all([ Blog.insertOne({...}), // 1000ms User.UpdateOne({...}) // 1000ms ]) } // 실행시간 : 1000ms // 질문코드 async createBlog() => { const blogPromise = Blog.insertOne({...}) // 1000 ms const userPromise = User.updateOne({...}) // 1000 ms A(await blogPromise, await userPromise) } // 실행시간 : 1000ms(?)
미해결
Airflow 마스터 클래스
안녕하세요, 강의내용대로 다른 데스트탑에서 airflow docker 까지 다 작동이 되는데 localhost:8080 에서 사용자이름과 비밀번호가 'airflow'로 인증이 안되는데요 어떻게 하지요?
미해결
카프카 완벽 가이드 - 커넥트(Connect) 편
안녕하세요.. kafkaconnect 실습 수행 중 mysql_jdbc_oc_sink_orders_datetime_tab_02.json 문장안에 끝 부분에 "," 가 수정이 아직 안되어 있습니다. 깜빡 하신거 같네요..ㅎㅎ
미해결
카프카 완벽 가이드 - ksqlDB
안녕하세요, join query를 보면 left join이라고 작성한 경우 outer join으로 적용되고, 그냥 join은 inner join으로 적용되는 것 같은데 ksqldb의 기본 join은 무엇인가요?
미해결
카프카 완벽 가이드 - ksqlDB
안녕하세요, 만약 group by의 경우 일반 컬럼을 지정해서 리파티션이 되었다면 해당 source topic 레코드를 조회하면 그 때부터 key값에 group by로 지정한 일반 컬럼이 key로 나오게 되는건가요? 아니면 group by 만을 위한 리파티션 수행 내부 토픽이 별도로 생성되는거고 source 토픽은 그냥 유지되는건가요?
미해결
15일간의 빅데이터 파일럿 프로젝트
안녕하세요! 강의 듣다 궁금한 점이 몇 가지 있어 질문 드립니다. 먼저 강의에서는 빅디님께서 각 서버에 설치할 프로그램을 알려주셔서 편하게 설치는 했는데, 어떤 기준으로 설치할 프로그램을 선택하고 각 서버에 설치할 프로그램을 나눠서 아키텍처를 짜셨는지 빅디님의 관점?이 궁금합니다. 예를 들어 postgreSQL는 서버 1에만 설치하고, HBase Region 같은 경우는 서버 세 곳 모두 설치 한 이유와, 다른 RDBMS 중에서도 postgreSQL을 선택한 특별한 이유 같은거요..! 그리고 서비스 중간에 서버를 늘리려고 할 때 추가해야 하는 서버 수는 어떻게 정하나요? 모니터링 하다가 서버 전체 메모리의 몇 퍼센트를 차지하게 되면 서버를 늘려야 한다 이런 기준이 있을까요? 비용은 제외하고 기술적인 부분에서 기준으로 세울만한 건 어떤게 있는지 궁금합니다. 현업에서는 프로젝트 특성마다 다 다르게 설계를 해야 할 테고 3V 관점으로 봐야 한다는 건 알겠는데 조금 더 구체적인 예시가 있으면 이해하는데 도움이 많이 될 것 같습니다! 감사합니다:]
미해결
[아파치 카프카 애플리케이션 프로그래밍] 개념부터 컨슈머, 프로듀서, 커넥트, 스트림즈까지!
카프카 인강을 듣고 처음 적용하려고 하고 있는데요. 혹시 프로듀서에서 컨슈머로 데이터를 받고, API 통신을 4회 정도 하여 DB에 넣는다고 하면 , 적정하지 않은 컨슈머 렉으로 진행되는 부분이 있지 않을까 고민이 되어서요. 비동기 방식으로 진행하고 큐에 데이터를 쌓고 진행하려고 하는데, 큐에 데이터가 많이 쌓인다면 메모리 이슈로도 진행될 것 같아서요. 요점은 프로듀서에서 컨슈머 로 데이터가 들어오면 비동기 방식으로 큐에 쌓은다. 큐에 쌓은 데이터를 API 통신으로 데이터를 가공한 후 DB에 넣어준다. 이렇게 되면 카프카에서 컨슈머로 렉이 발생되지는 않겠지만 큐에 데이터가 쌓이면서 메모리 문제가 되지 않을까 고민이 있습니다.
미해결
카프카 완벽 가이드 - ksqlDB
안녕하세요, ksqldb에는 RDBMS와 달리 datetime 타입이 따로 없는 것 같은데요, timestamp 타입은 타임존이 반영된 타입인건가요?
미해결
Airflow 마스터 클래스
전체 에러 로그: 8fbc4f9f281a *** Found local files: *** * /opt/airflow/logs/dag_id=dags_python_with_custom_hook_bulk_load/run_id=manual__2024-03-06T03:55:34.102649+00:00/task_id=insrt_postgres/attempt=1.log [2024-03-06T03:55:35.452+0000] {taskinstance.py:1979} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: dags_python_with_custom_hook_bulk_load.insrt_postgres manual__2024-03-06T03:55:34.102649+00:00 [queued]> [2024-03-06T03:55:35.456+0000] {taskinstance.py:1979} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: dags_python_with_custom_hook_bulk_load.insrt_postgres manual__2024-03-06T03:55:34.102649+00:00 [queued]> [2024-03-06T03:55:35.456+0000] {taskinstance.py:2193} INFO - Starting attempt 1 of 1 [2024-03-06T03:55:35.464+0000] {taskinstance.py:2214} INFO - Executing <Task(PythonOperator): insrt_postgres> on 2024-03-06 03:55:34.102649+00:00 [2024-03-06T03:55:35.469+0000] {standard_task_runner.py:60} INFO - Started process 262 to run task [2024-03-06T03:55:35.471+0000] {standard_task_runner.py:87} INFO - Running: ['***', 'tasks', 'run', 'dags_python_with_custom_hook_bulk_load', 'insrt_postgres', 'manual__2024-03-06T03:55:34.102649+00:00', '--job-id', '620', '--raw', '--subdir', 'DAGS_FOLDER/dags_python_with custom_hook_bulk_load.py', '--cfg-path', '/tmp/tmp0zav_1t7'] [2024-03-06T03:55:35.472+0000] {standard_task_runner.py:88} INFO - Job 620: Subtask insrt_postgres [2024-03-06T03:55:35.504+0000] {task_command.py:423} INFO - Running <TaskInstance: dags_python_with_custom_hook_bulk_load.insrt_postgres manual__2024-03-06T03:55:34.102649+00:00 [running]> on host 8fbc4f9f281a [2024-03-06T03:55:35.555+0000] {taskinstance.py:2510} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='***' AIRFLOW_CTX_DAG_ID='dags_python_with_custom_hook_bulk_load' AIRFLOW_CTX_TASK_ID='insrt_postgres' AIRFLOW_CTX_EXECUTION_DATE='2024-03-06T03:55:34.102649+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2024-03-06T03:55:34.102649+00:00' [2024-03-06T03:55:35.556+0000] {custom_postgres_hook.py:24} INFO - 적재 대상 파일: /opt/***/files/TbCorona19CountStatus/20240306/TbCorona19CountStatus.csv [2024-03-06T03:55:35.556+0000] {custom_postgres_hook.py:25} INFO - 테이블 :TbCorona19CountStatus_bulk2 [2024-03-06T03:55:35.562+0000] {base.py:83} INFO - Using connection ID 'conn-db-postgres-custom' for task execution. [2024-03-06T03:55:35.582+0000] {custom_postgres_hook.py:34} INFO - TbCorona19CountStatus_bulk2.S_DT: 개행문자 제거 [2024-03-06T03:55:35.583+0000] {custom_postgres_hook.py:34} INFO - TbCorona19CountStatus_bulk2.T_DT: 개행문자 제거 [2024-03-06T03:55:35.583+0000] {custom_postgres_hook.py:38} INFO - 적재 건수:1212 [2024-03-06T03:55:35.583+0000] {custom_postgres_hook.py:40} INFO - postgresql://***:***@172.28.0.3/*** [2024-03-06T03:55:35.587+0000] {warnings.py:110} WARNING - /opt/***/plugins/hooks/custom_postgres_hook.py:43: UserWarning: pandas only supports SQLAlchemy connectable (engine/connection) or database string URI or sqlite3 DBAPI2 connection. Other DBAPI2 objects are not tested. Please consider using SQLAlchemy. file_df.to_sql(name = table_name, [2024-03-06T03:55:35.589+0000] {taskinstance.py:2728} ERROR - Task failed with exception Traceback (most recent call last): File "/home/airflow/.local/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 444, in execute task result = execute callable(context=context, **execute_callable_kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 414, in execute callable return execute_callable(context=context, **execute_callable_kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.11/site-packages/airflow/operators/python.py", line 200, in execute return_value = self.execute_callable() ^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.11/site-packages/airflow/operators/python.py", line 217, in execute_callable return self.python_callable(*self.op_args, **self.op_kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/opt/airflow/dags/dags_python_with custom_hook_bulk_load.py", line 14, in insrt_postgres custom_postgres_hook.bulk_load(table_name=tbl_nm, file_name=file_nm, delimiter=',', File "/opt/airflow/plugins/hooks/custom_postgres_hook.py", line 43, in bulk_load file_df.to_sql(name = table_name, File "/home/airflow/.local/lib/python3.11/site-packages/pandas/util/_decorators.py", line 333, in wrapper return func(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.11/site-packages/pandas/core/generic.py", line 3084, in to_sql return sql.to_sql( ^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.11/site-packages/pandas/io/sql.py", line 842, in to_sql return pandas_sql.to_sql( ^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.11/site-packages/pandas/io/sql.py", line 2848, in to_sql table.create() File "/home/airflow/.local/lib/python3.11/site-packages/pandas/io/sql.py", line 984, in create if self.exists(): ^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.11/site-packages/pandas/io/sql.py", line 970, in exists return self.pd_sql.has_table(self.name, self.schema) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.11/site-packages/pandas/io/sql.py", line 2863, in has_table return len(self.execute(query, [name]).fetchall()) > 0 ^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/home/airflow/.local/lib/python3.11/site-packages/pandas/io/sql.py", line 2670, in execute cur = self.con.cursor() ^^^^^^^^^^^^^^^ AttributeError: 'Engine' object has no attribute 'cursor' [2024-03-06T03:55:35.597+0000] {taskinstance.py:1149} INFO - Marking task as FAILED. dag_id=dags_python_with_custom_hook_bulk_load, task_id=insrt_postgres, execution_date=20240306T035534, start_date=20240306T035535, end_date=20240306T035535 [2024-03-06T03:55:35.605+0000] {standard_task_runner.py:107} ERROR - Failed to execute job 620 for task insrt_postgres ('Engine' object has no attribute 'cursor'; 262) [2024-03-06T03:55:35.643+0000] {local_task_job_runner.py:234} INFO - Task exited with return code 1 [2024-03-06T03:55:35.655+0000] {taskinstance.py:3309} INFO - 0 downstream tasks scheduled from follow-on schedule check 결과로 cursor가 attribute로 없다는 오류가 발생합니다. 기존에 있던 이후 글은 글자수 제한으로 삭제합니다.
해결됨
Airflow 마스터 클래스
알찬 강의 항상 감사드립니다. 작업 위치1 (vsc상에서 작성하고 git push를 하는 위치) : [~/Desktop/airflow강의] 작업 위치2 (WSL에서 git pull을 하는 위치) : [~/airflow] 저는 위와 같은 경로상에서 수업 실습을 진행하고 있으며, '외부 파이썬 함수 수행하기' 강의에서 .env 파일을 다음과 같이 작성했습니다. WORKSPACE_FOLDER=/Users/hpsm5187/Desktop/airflow강의 airflow가 인식하는 작업 폴더는 '작업 위치2'인 걸로 아는데, .env파일 내의 경로는 '작업 위치1'의 경로임에도 이상 없이 작동이 되는 부분이 의아합니다. .env파일을 git에서 인식하지 않도록 처리했는데 .env에서 정의해준 경로는 어떻게 반영이 되는건가요?
미해결
다양한 사례로 익히는 SQL 데이터 분석
안녕하세요? 월단위 카테고리별 매출액과 주문건수 및 전체매출액 대비 비율 sql로 구하기 강의 에서, 하기 함수에 distinct를 반드시 사용해야만 하는 이유를 잘 모르겠습니다. order_date에 order_id가 여러 번 있을 수 있는 상황은 전혀 없을까요? 답변 부탁드립니다.
해결됨
Airflow 마스터 클래스
안녕하세요 강사님, 문제 사항 질문 드립니다. Bash Operator DAG만들기 부분에서 docker_compose.yaml파일을 vi편집기로 열어보았는데요, 이렇게 표시되며 j, k로 스크롤해보아도 변함이 없습니다. 혹시 컴포즈 파일이 비어있는 상태인걸까요? docker compose up명령어는 문제 없이 실행됩니다. ++ 질문 수정 vscode로 파일을 확인해보니 같은 파일이 두 개 있고, 하나는 비어있고 하나는 강의에서와 동일하게 내용이 차 있습니다. ++ 해결되었습니다. 찾아보니 새로 생긴 파일 하나는 macos자체에서 생성하는 메타데이터 파일이라고 하는것 같네요. 시간 좀 지나고 다시 vi편집기로 열어보니 내용이 있는 파일이 열렸습니다. 감사합니다!
미해결
다양한 사례로 익히는 SQL 데이터 분석
안녕하세요. 강의 잘 듣고 있는데요 with 절이 계속 사용하고 있는것 같은데 with 절 말고 서브쿼리로 작성해도 상관없나요? 그리고 상관없다면 with절하고 서브쿼리 중에서 어떤 쿼리가 더 성능이 좋은 쿼리인가요?
해결됨
카프카 완벽 가이드 - 커넥트(Connect) 편
안녕하세요, 질문이 있습니다. auto.evolve=true로 sink 쪽에 설정하고 source DB의 칼럼 정보를 업데이트 한 후에 레코드를 삽입하지 않으면 토픽에 메시지가 전송되지 않으니 target DB에는 반영되지 않는건가요?
미해결
[리뉴얼] 처음하는 SQL과 데이터베이스(MySQL) 부트캠프 [입문부터 활용까지]
https://davelee-fun.github.io/ 이 사이트 크롤링을 1번 사진처럼 4파트로 분리해서 크롤링후 2번 사진 처럼 만들어놓은 DB 테이블에 insert하는 코드를 작성해보았는데, (3번사진) 크롤링부분에서 잘못된건지 코드가 잘 돌아가지 않습니다. 혹시 어느부분이 잘못됬는지 알수있을까요? import requests from bs4 import BeautifulSoup import re pg=1 if pg==1: #1페이지 크롤링시 res = requests.get('https://davelee-fun.github.io/') #웹사이트 get request 1페이지 soup = BeautifulSoup(res.content, 'html.parser') #html 파싱 items = soup.select('div.card h') product_id=17890001 #promary key 임의설정 page_num=pg for item in items: recom_info=item.select_one('a.text-dark') #추천 product_name=item.select_one('h4.card-text') #상품명 provider=item.select_one('a[target="_blank"]') #제공자 -> teddy register_date=item.select_one('span.post-date') #등록일자 -> 05 jun 2020 SQL = """INSERT INTO product_info VALUES('"""+product_id+"""', '""" + str(recom_info.get_text()) + """','""" + str(product_name.get_text()) + """','""" + str(provider.get_text()) + """', '""" + str(register_date.get_text()) + """','""" +page_num+ """'); """ print(SQL) cursor.execute(SQL) 우선 1페이지를 크롤링한 값을 넣어보려고 했습니다.
미해결
카프카 완벽 가이드 - 코어편
똑같이 아이피 고정해서 했는데 kafka < 라는걸 계속 찾네요 ㅠㅠ192.xx 대역으로 하면 못찾습니다.. 그래서일단은 windows hosts 설정에 kafka 도메인으로 고정IP를 넣어줘서 해결은 했는데..도커로 띄워서 그런걸까요?
