안녕하세요,강의 내용 중 table에 대해서 source를 topic으로 하는 경우 pull 쿼리가 불가능 한 이유에 대해 강사님께서 구체적으로 확인이 되지 않았지만 rocksdb 이슈로 추정된다라고 말씀하신 것 같은데요, 혹시 아래의 이유가 아닐 지 확인 부탁드립니다. https://docs.ksqldb.io/en/latest/concepts/materialized-views/In ksqlDB, a table can be materialized into a view or not. If a table is created directly on top of a Kafka topic, it's not materialized. Non-materialized tables can't be queried, because they would be highly inefficient. On the other hand, if a table is derived from another collection, ksqlDB materializes its results, and you can make queries against it. document에 따르면 source가 topic인 table의 경우는 mview가 아니라고 하는 것 같습니다.(mview가 아니기 때문에 비효율적이다.) 그렇다면 이 경우는 rocksdb를 통해 stateful한 결과를 저장한 뒤 가져오는 것이 아닌 토픽으로부터 전체 레코드를 읽어와서 compact한 처리를 하는 케이스라고 볼 수 있을 것 같습니다. 즉, source가 topic인 케이스는 오히려 rocksdb를 사용하지 않는 케이스이기 때문에 그런 것이 아닐까 추측이 되는데요 어떻게 생각하시는지 궁금합니다.   

안녕하세요강의에서 xcom_pull 메서드 사용시 task_ids를 지정하지 않으면 가장 마지막 태스크의 키값을 가져온다고 하셨는데요, 실습코드를 돌려보니 먼저 실행되었던 태스크의 밸류값을 리턴합니다. from airflow import DAG import pendulum from airflow.decorators import task with DAG( dag_id="dags_python_with_xcom_eg1", schedule="30 6 * * *", start_date=pendulum.datetime(2024, 3, 1, tz="Asia/Seoul"), catchup=False, ) as dag: @task(task_id="python_xcom_push_task1") def xcom_push1(**kwargs): ti = kwargs["ti"] ti.xcom_push(key="result1", value="value_1") ti.xcom_push(key="result2", value=[1, 2, 3, 4]) @task(task_id="python_xcom_push_task2") def xcom_push2(**kwargs): ti = kwargs["ti"] ti.xcom_push(key="reuslt1", value="value_2") ti.xcom_push(key="reuslt2", value=[1, 2, 3]) @task(task_id="python_xcom_pull_task") def xcom_pull(**kwargs): ti = kwargs["ti"] value1 = ti.xcom_pull(key="result2") # [1, 2, 3] value2 = ti.xcom_pull( key="result1", task_ids="python_xcom_push_task1" ) # value_1 print(value1) print(value2) xcom_push1() >> xcom_push2() >> xcom_pull() dag는 위와 같이 작성했고요, [1, 2, 3]이 반환될 것으로 기대했던 부분에서 [1, 2, 3, 4]가 반환되어서 혹시 제가 코드를 잘못 작성한건지 궁금해서 여쭤봅니다 

안녕하세요 강사님!dags_python_with_templates.py 파일에서 show_templates 함수를 아무 인자도 넣지 않고 실행을 시키는데요, dag 실행시 출력되는 키워드 인자들은 에어플로우 태스크 객체에 기본으로 입력되는 값들인가요? 정확한 원리가 궁금하여 여쭤봅니다

안녕하세요 얄코님 좋은 강의 덕분에 하루 하루 성장해가는 1인 입니다혹시 인덱스, 파티셔닝 이런 용어들이 있어 찾아봐도 이해가 잘 되지 않더라구요추 후에 샤딩?, 레플리케이션?, 파티셔닝, 인덱스등 DB 튜닝? 관련하여 강의를 출시할 예정이 있으신가요?   

3정규화용주문 테이블에주문번호상품번호상품명있으면 3정규화 대상으로 주문 테이블은주문번호상품번호 상품 테이블 따로 만들어서상품번호상품명 이렇게 따로 관리해야한다는거같은데용문제가요,상품명이 바뀌게 되면주문 테이블에 상품명도 바껴서예전에 주문했던 상품명이 달라져서 혼란스러울것같은데이런거는 3정규화 대상인가요정책에 따라서 달라지나요

powershell에서 wsl --install을 입력한 후에ubuntu 시작하는중... 이 출력되고그 다음에 응답이 없는데이런 경우 어떻게 해야할까요?

안녕하세요.  5개월쯤 완강하고 실습 했을때는 문제가 없었는데,지금 다시 마지막 강의와 같이 배포하려고 하니 github action에서 Node.js 16 actions are deprecated. Please update the following actions to use Node.js 20: actions/checkout@v3. For more information see: https://github.blog/changelog/2023-09-22-github-actions-transitioning-from-node-16-to-node-20/.다음과 같은 에러가 납니다. 혹시 어떤 문제인지 알 수 있을까요?

안녕하세요, 늘 양질의 강의 제공해 주심에 감사합니다.강의를 듣다가 알쏭달쏭한 부분이 있어 질문드립니다. Schema Registry 를 활용하여 하위(Backward) 호환성 유지 시, 새로운 스키마를 설정되게 되면 Worker Conenct 를 강제로 실행시켜 줘서 Sink Connector 가 새로운 스키마가 갱신되도록 해줘야하는 절차가 꼭 필요한가요? (Consumer 가 새로운 버전으로 스키마 업데이트가 되도록 강제하기 위한 수동 절차가 꼭 필요한 것인지 궁금합니다. 아님 이러한 부분도 자동으로 Schema registry 에서 Sink 쪽으로 콜백을 주어 캐시된 스키마에 대한 정보 업데이트가 되도록 되는건지,,) (Forward 호환성인 경우) Source Connector 에서 V1 버전 스키마로 메세지를 발행하다가, 어느 순간 신규 스키마 버전 V2 로 메세지를 발행하게 되어 Schema registry 에 V2 가 등록이 된 상태에서, Sink 쪽에서는 이 메세지를 받게 되면 schema id : v2 인 정보를 카프카 메세지 내에 정보를 통해 알게 될텐데, 그럼에도 아직 워커 커넥트 내부 캐시된 스키마 정보엔 v2 가 없더라도, Forward 호환성인 경우에는 Sink Connector(Consumer) 입장에선 Schema registry 를 새로이 호출하여 스키마 업데이트를 하지 않는걸까요 ?  보통 운영환경에서 스키마 관리 자체를 DB source connector 에서 자동으로 만들어주는 스키마가 아니라 Schema registry 에 직접 새로운 스키마를 등록하여 운영환경에서 활용하시는지, 혹은 SOurce connector 의 자동 스키마 완성으로 사용하시는지 궁금 합니다.  양질의 강의에 다시한번 감사드리며 답변 부탁드리겠습니다. 좋은 하루 되세요.감사합니다. 

안녕하세요 선생님실습에서 multipart-topic으로 key 없이 전송을 했는데 모조리 partition2로 가는데 어떤 설정을 살펴보면 될까요?kafka는 confluent 7.6으로 실습하고 있습니다.key값을 준 경우에는 강의 실습과 동일하게 진행되었습니다.

안녕하세요 KStreamJoinKTable 실행시 에러가 발생해서요 ㅠ 에러는 아래와 같습니다.  [order-join-application-517e77af-95a4-4398-9800-1cef92f224d1-StreamThread-1] ERROR org.apache.kafka.streams.KafkaStreams - stream-client [order-join-application-517e77af-95a4-4398-9800-1cef92f224d1] All stream threads have died. The instance will be in error state and should be closed. [order-join-application-517e77af-95a4-4398-9800-1cef92f224d1-StreamThread-1] INFO org.apache.kafka.streams.processor.internals.StreamThread - stream-thread [order-join-application-517e77af-95a4-4398-9800-1cef92f224d1-StreamThread-1] Shutdown complete Exception in thread "order-join-application-517e77af-95a4-4398-9800-1cef92f224d1-StreamThread-1" java.lang.UnsatisfiedLinkError: /private/var/folders/x8/qq1k6wyn1qs76952l1d1t0qh0000gn/T/librocksdbjni11881960449213041495.jnilib: dlopen(/private/var/folders/x8/qq1k6wyn1qs76952l1d1t0qh0000gn/T/librocksdbjni11881960449213041495.jnilib, 0x0001): tried: '/private/var/folders/x8/qq1k6wyn1qs76952l1d1t0qh0000gn/T/librocksdbjni11881960449213041495.jnilib' (mach-o file, but is an incompatible architecture (have 'x86_64', need 'arm64e' or 'arm64')), '/System/Volumes/Preboot/Cryptexes/OS/private/var/folders/x8/qq1k6wyn1qs76952l1d1t0qh0000gn/T/librocksdbjni11881960449213041495.jnilib' (no such file), '/private/var/folders/x8/qq1k6wyn1qs76952l1d1t0qh0000gn/T/librocksdbjni11881960449213041495.jnilib' (mach-o file, but is an incompatible architecture (have 'x86_64', need 'arm64e' or 'arm64')) at java.base/jdk.internal.loader.NativeLibraries.load(Native Method) at java.base/jdk.internal.loader.NativeLibraries$NativeLibraryImpl.open(NativeLibraries.java:388) at java.base/jdk.internal.loader.NativeLibraries.loadLibrary(NativeLibraries.java:232) at java.base/jdk.internal.loader.NativeLibraries.loadLibrary(NativeLibraries.java:174) at java.base/java.lang.ClassLoader.loadLibrary(ClassLoader.java:2389) at java.base/java.lang.Runtime.load0(Runtime.java:755) at java.base/java.lang.System.load(System.java:1953) at org.rocksdb.NativeLibraryLoader.loadLibraryFromJar(NativeLibraryLoader.java:78) at org.rocksdb.NativeLibraryLoader.loadLibrary(NativeLibraryLoader.java:56) at org.rocksdb.RocksDB.loadLibrary(RocksDB.java:64) at org.rocksdb.RocksDB.<clinit>(RocksDB.java:35) at org.rocksdb.DBOptions.<clinit>(DBOptions.java:21) at org.apache.kafka.streams.state.internals.RocksDBStore.openDB(RocksDBStore.java:133) at org.apache.kafka.streams.state.internals.RocksDBStore.init(RocksDBStore.java:229) at org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48) at org.apache.kafka.streams.state.internals.ChangeLoggingKeyValueBytesStore.init(ChangeLoggingKeyValueBytesStore.java:44) at org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48) at org.apache.kafka.streams.state.internals.CachingKeyValueStore.init(CachingKeyValueStore.java:58) at org.apache.kafka.streams.state.internals.WrappedStateStore.init(WrappedStateStore.java:48) at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.lambda$init$0(MeteredKeyValueStore.java:101) at org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:806) at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.init(MeteredKeyValueStore.java:101) at org.apache.kafka.streams.processor.internals.AbstractTask.registerStateStores(AbstractTask.java:210) at org.apache.kafka.streams.processor.internals.StreamTask.initializeStateStores(StreamTask.java:275) at org.apache.kafka.streams.processor.internals.AssignedTasks.initializeNewTasks(AssignedTasks.java:76) at org.apache.kafka.streams.processor.internals.TaskManager.updateNewAndRestoringTasks(TaskManager.java:397) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:779) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:697) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:670) - 카프카를 공부하시면서 생긴 질문들을 남겨주세요. 상세히 작성하면 더 좋아요! - 먼저 유사한 질문이 있었는지 검색해보세요. - 서로 예의를 지키며 존중하는 문화를 만들어가요. - 잠깐! 인프런 서비스 운영 관련 문의는 1:1 문의하기를 이용해주세요.

안녕하세요ssh master02 입력시 permission denied 에러가 발생했습니다. ㅠㅠ 이전 강의에서 정상적으로 SSH 등록했고 AMI 이미지 만들고 launch instance까지 완료했는데 오류가 발생했는데요.  해결 방법이 궁금합니다

안녕하세요. 카프카를 실무 도입전 공부를 하고 있습니다. 강의에서 3버전대는 아직 실무에 도입한 사례가 없다고 하시는데 강의가 2년전이라서 요즈음에도 아직 3버전이 크게 상용화가 안되었나요? 첫 도입 하려고 하다보니 신경쓸게 많아 질문 드립니다.감사합니다

AMI 생성 후에 instance를 각각 4개씩 launch를 했는데요영상 속 이미지와 다르게 sudo vim etc/hosts를 해도 빈 파일이 열립니다.ip 리스트를 작성하고 저장하려고 하니 can't open 에러가 발생하고요. 등록한 ip 리스트를 이용해서 노드 간 통신을 하는 것으로 보여지는데요.무시하고 진행해도 되는건가요?

OperationalError: (1205, 'Lock wait timeout exceeded; try restarting transaction')해당 오류가 발생합니다. 아래는 제 소스코드입니다사실 수업과 다를 바는 없다고 생각은 드는데..다른 구문은 다 정상작동하는데 UPDATE 구문만 작동을 안하는데 이유가 있을까요? 전체 오류 메시지도 사진으로 첨부합니다 Workbench에서도 동일한 오류가 발생합니다 ㅠㅠㅠㅠ #1. 라이브러리 importimport pymysql#2. 데이터베이스에 접속하기## host, port 등등은 mysql에서 확이db = pymysql.connect(host='localhost',port=3306,user='root',passwd='비밀번호입력',db='ecommerce',charset='utf8')#3. 커서 가져오기cursor = db.cursor()#4.sql 구문 만들기(INSERT, CREATE, DELETE 등등)SQL = """UPDATE product SETTITLE ='달리샵린넨원피스 뷔스티에 썸머 가디건 코디전',ORI_PRICE=33000,DISCOUNT_PRICE=9900,DISCOUNT_PERCENT=70WHERE PRODUCT_CODE = '215673141'""" #5 SQL 구문 실행하기cursor.execute(SQL) #6 DB에 Complete 하기db.commit()#7 DB 종료하기db.close() 

섹션 1 docker 설치 단계에서sudo service docker start 입력하면 Starting Docker: docker [OK] 라고 나옵니다.하지만 sudo docker run hello-world 입력 하면 docker daemon이 연결되지 않는다고 나옵니다.sudo service docker status 입력해서 확인해 보니 Docker is not running 상태로 지속되고 있습니다.sudo service docker restart 입력하고 해도 동일한 현상이 나옵니다./var/log/docker.log를 열어 확인해보니 아래에 나와있듯이 plugin 로딩 시 에러가 나오는 것 같습니다. 애초에 알려주신 https://docs.docker.com/engine/install/ubuntu 링크의 docker 설치 명령어가 지난 몇달 사이에 바뀌었는지, 강의영상에서의 설치 명령어와 약간 달랐습니다.이 링크에서의 명령어를 사용하지 않고 강의영상에서의 명령어로 입력하면 문제가 해결될까요?아니면 다른 방법이 있는지 문의드립니다. 

자동 커밋 옵션을 활성화한 경우 일정 시간마다 오프셋 커밋을 실행하는 것을 이해했습니다. 결제 이벤트를 처리하는 카프카 컨슈머가 100건의 레코드를 가져와서 한번에 처리하도록 구현되어 있을 때 레코드 처리 중간에 장애가 발생하여 오프셋 커밋을 하지 못하는 경우 어떤 일이 일어나는지 궁금합니다. 50번째 레코드를 처리하다가 장애가 발생했다고 가정하면 오프셋 커밋이 되지 못했으므로 다른 컨슈머 애플리케이션이 이미 처리된 100건의 레코드를 다시 가져와서 1~50번째 레코드가 중복으로 이벤트가 처리될 것 같습니다. 이는 컨슈머 애플리케이션은 언제든지 중복으로 레코드를 소비할 수 있다는 것으로 이해가 되고, 컨슈머 애플리케이션 로직을 작성할 때는 항상 중복 레코드가 처리되지 않도록 방어해야한다는 것을 의미한다고 생각합니다. 제가 만약 방어 로직을 작성한다고 하면 레코드의 메시지마다 유일한 키값을 같이 전송하여 해당 값을 Redis에 저장하고, redis에 저장된 상태라면 메시지를 처리하지 않도록 구현할 것 같습니다. 제가 오프셋과 관련하여 이해한 내용이 정확한지, 그리고 실무에서도 Redis를 활용한 중복 처리 방지 로직이 효과적인 방법인지 궁금합니다. 또한, 이와 같은 상황을 해결하기 위해 다른 방법을 사용하시는지도 알고 싶습니다. 답변 주시면 감사하겠습니다! 

수업자료 MARIADB설치PPT - 10.7.3버전이 없네요구글링 10.7.3 다운로드는 Mirror가 Germany 뿐이구 

Task 실패시 email 발송하기sla 로 task 수행현황 감시하고 email 발송하기 이 두개의 영상이 바뀐것 같아 보입니다 확인 해봐주세요~

안녕하세요 강의 잘 보고 있습니다.복제강의 수강중 궁금한점이리더 - 팔로워 간의 복제시 팔로워에서 리더의 오프셋을 확인해 차이점이 있다면 팔로워로 저장한다고 이해했는데요.이러한 복제 과정으로 인해 리더파티션에 이슈가 생길만한 케이스가 있는지 궁금합니다.기본적으로 리더 파티션이 프로듀서 - 컨슈머와의 통신을 담당하기에 팔로워 파티션에서 복제과정을 담당하게 되는 형태인것 같아 문제가 없을것 같긴한데 .. 꼭 이렇게 생각하고 넘어가면 문제가 생기더라구요 ㅎ..혹시 복제과정이 리더 파티션 성능에 영향을 주게될만한 이슈가 있을지 우려되서 질문드립니다

안녕하세요! 2강에서 설명해주신 레플리케이션 설정을 통한 고가용성 외에 서버 랙, 리전 장애에도 대비할 수 있는 카프카 설정이 무엇인지 궁금하여 질문드립니다.