(아나콘다를 깔고나서 새 파일에서 뭘 수정하려고하면 바로 연결이 끊기는 문제가 생겨 주피터 노트북으로 설치하였습니다. 필요한 라이브러리들은 모두 따로 설치하였습니다.)버전은 아래와 같습니다.오류가 발생하는 부분은 이 부분인데query문이 있는 셀의 전체 오류문은 아래와 같습니다--------------------------------------------------------------------------- UnicodeDecodeError Traceback (most recent call last) Cell In[12], line 4 1 query = """ 2 select * from nw.customers 3 """ ----> 4 df = pd.read_sql_query(sql=query, con=postgres_engine) 5 df.head(10) File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\pandas\io\sql.py:485, in read_sql_query(sql, con, index_col, coerce_float, params, parse_dates, chunksize, dtype, dtype_backend) 482 dtype_backend = "numpy" # type: ignore[assignment] 483 assert dtype_backend is not lib.no_default --> 485 with pandasSQL_builder(con) as pandas_sql: 486 return pandas_sql.read_query( 487 sql, 488 index_col=index_col, (...) 494 dtype_backend=dtype_backend, 495 ) File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\pandas\io\sql.py:851, in pandasSQL_builder(con, schema, need_transaction) 848 raise ImportError("Using URI string without sqlalchemy installed.") 850 if sqlalchemy is not None and isinstance(con, (str, sqlalchemy.engine.Connectable)): --> 851 return SQLDatabase(con, schema, need_transaction) 853 warnings.warn( 854 "pandas only supports SQLAlchemy connectable (engine/connection) or " 855 "database string URI or sqlite3 DBAPI2 connection. Other DBAPI2 " (...) 858 stacklevel=find_stack_level(), 859 ) 860 return SQLiteDatabase(con) File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\pandas\io\sql.py:1576, in SQLDatabase.__init__(self, con, schema, need_transaction) 1574 self.exit_stack.callback(con.dispose) 1575 if isinstance(con, Engine): -> 1576 con = self.exit_stack.enter_context(con.connect()) 1577 if need_transaction and not con.in_transaction(): 1578 self.exit_stack.enter_context(con.begin()) File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\base.py:3268, in Engine.connect(self) 3245 def connect(self) -> Connection: 3246 """Return a new :class:`_engine.Connection` object. 3247 3248 The :class:`_engine.Connection` acts as a Python context manager, so (...) 3265 3266 """ -> 3268 return self._connection_cls(self) File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\base.py:145, in Connection.__init__(self, engine, connection, _has_events, _allow_revalidate, _allow_autobegin) 143 if connection is None: 144 try: --> 145 self._dbapi_connection = engine.raw_connection() 146 except dialect.loaded_dbapi.Error as err: 147 Connection._handle_dbapi_exception_noconnection( 148 err, dialect, engine 149 ) File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\base.py:3292, in Engine.raw_connection(self) 3270 def raw_connection(self) -> PoolProxiedConnection: 3271 """Return a "raw" DBAPI connection from the connection pool. 3272 3273 The returned object is a proxied version of the DBAPI (...) 3290 3291 """ -> 3292 return self.pool.connect() File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py:452, in Pool.connect(self) 444 def connect(self) -> PoolProxiedConnection: 445 """Return a DBAPI connection from the pool. 446 447 The connection is instrumented such that when its (...) 450 451 """ --> 452 return _ConnectionFairy._checkout(self) File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py:1269, in _ConnectionFairy._checkout(cls, pool, threadconns, fairy) 1261 @classmethod 1262 def _checkout( 1263 cls, (...) 1266 fairy: Optional[_ConnectionFairy] = None, 1267 ) -> _ConnectionFairy: 1268 if not fairy: -> 1269 fairy = _ConnectionRecord.checkout(pool) 1271 if threadconns is not None: 1272 threadconns.current = weakref.ref(fairy) File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py:716, in _ConnectionRecord.checkout(cls, pool) 714 rec = cast(_ConnectionRecord, pool._do_get()) 715 else: --> 716 rec = pool._do_get() 718 try: 719 dbapi_connection = rec.get_connection() File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\impl.py:169, in QueuePool._do_get(self) 167 return self._create_connection() 168 except: --> 169 with util.safe_reraise(): 170 self._dec_overflow() 171 raise File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\util\langhelpers.py:146, in safe_reraise.__exit__(self, type_, value, traceback) 144 assert exc_value is not None 145 self._exc_info = None # remove potential circular references --> 146 raise exc_value.with_traceback(exc_tb) 147 else: 148 self._exc_info = None # remove potential circular references File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\impl.py:167, in QueuePool._do_get(self) 165 if self._inc_overflow(): 166 try: --> 167 return self._create_connection() 168 except: 169 with util.safe_reraise(): File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py:393, in Pool._create_connection(self) 390 def _create_connection(self) -> ConnectionPoolEntry: 391 """Called by subclasses to create a new ConnectionRecord.""" --> 393 return _ConnectionRecord(self) File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py:678, in _ConnectionRecord.__init__(self, pool, connect) 676 self.__pool = pool 677 if connect: --> 678 self.__connect() 679 self.finalize_callback = deque() File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py:902, in _ConnectionRecord.__connect(self) 900 self.fresh = True 901 except BaseException as e: --> 902 with util.safe_reraise(): 903 pool.logger.debug("Error on connect(): %s", e) 904 else: 905 # in SQLAlchemy 1.4 the first_connect event is not used by 906 # the engine, so this will usually not be set File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\util\langhelpers.py:146, in safe_reraise.__exit__(self, type_, value, traceback) 144 assert exc_value is not None 145 self._exc_info = None # remove potential circular references --> 146 raise exc_value.with_traceback(exc_tb) 147 else: 148 self._exc_info = None # remove potential circular references File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\pool\base.py:898, in _ConnectionRecord.__connect(self) 896 try: 897 self.starttime = time.time() --> 898 self.dbapi_connection = connection = pool._invoke_creator(self) 899 pool.logger.debug("Created new connection %r", connection) 900 self.fresh = True File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\create.py:637, in create_engine.<locals>.connect(connection_record) 634 if connection is not None: 635 return connection --> 637 return dialect.connect(*cargs, **cparams) File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\sqlalchemy\engine\default.py:616, in DefaultDialect.connect(self, *cargs, **cparams) 614 def connect(self, *cargs, **cparams): 615 # inherits the docstring from interfaces.Dialect.connect --> 616 return self.loaded_dbapi.connect(*cargs, **cparams) File ~\AppData\Local\Programs\Python\Python312\Lib\site-packages\psycopg2\__init__.py:122, in connect(dsn, connection_factory, cursor_factory, **kwargs) 119 kwasync['async_'] = kwargs.pop('async_') 121 dsn = _ext.make_dsn(dsn, **kwargs) --> 122 conn = _connect(dsn, connection_factory=connection_factory, **kwasync) 123 if cursor_factory is not None: 124 conn.cursor_factory = cursor_factory UnicodeDecodeError: 'utf-8' codec can't decode byte 0xb8 in position 63: invalid start byte 

안녕하세요 선생님 🙂 connection 정보를 새로운 환경에 이식 방법 질문드립니다.stackoverflow 등을 찾아보니 csv파일 또는 sh 파일로 구축하여 사용하는 것같은데요이 같은 방법은 보안에 취약할수있겠다는 생각이 들어서 어떤 방식으로 이식가능한 connection 정보를 저장하는것이 좋은지 질문드립니다. stackover reference - https://stackoverflow.com/questions/55626195/export-all-airflow-connections-to-new-environment즐거운 크리스마스 연휴되세요!감사합니다! 🙂

안녕하세요, 먼저 좋은 강의 잘 듣고 있습니다 (- -) (_ _)  컨슈머 랙 모니터링 - 처리량 이슈 파트에서 궁금한 점이 생겨서 질문드립니다. 파티션 1개 - 컨슈머 1개에서 파티션 2개 - 컨슈머 2개가 되면 linear하게 처리량이 늘어난다고 하셨는데, 어떻게 처리량이 늘어나는 건지 조금 이해가 안가서요. 상단과 같은 구조는 하나의 컨슈머가 0,1,2,3... 이렇게 하나씩 처리를 한다고 하면, 하단과 같은 구조는 다음과 같이 레코드를 동시에 소비하게 되어 2배로 늘어난다는 말씀일까요?컨슈머 1 - 0 : 2 : 4 : 6 : ...컨슈머 2 - 1 : 3 : 5 : 7 : ...( : 는 레코드가 소비되는 시점을 구분한 것 입니다!) 만약 그렇다면, offset은 어떻게 관리가 되는건지도 궁금하네요..

강사님, 안녕하세요! 강의 잘 듣고 있습니다.강의 들으면서 궁금한 사항이 생겨서 질문드려요.key를 지정해서 메시지를 전송하면 같은 key에 한해서는 같은 파티션으로 메시지가 전송되는 것을 보장 받을 수 있고, 해당 key로 전송된 메세지는 순서를 보장받을 수 있다. 까지는 이해했는데요~운영하면서 파티션의 개수가 늘어날 경우, 늘어난 직후에도 같은 key로 전송되는 메세지는 순서를 보장받을 수 있는 것일까요?(즉, 파티션 개수가 늘어나기 이전에 key:a, partition:0으로 전송되었다면 추후, 파티션 개수가 늘어나 리밸런싱이 일어나더라도 key:a 메세지는 partition:0으로 전송되는 것일까요?)

수업 강의 자료라는 것은 어느 메뉴로 들어가야하나요???? 메뉴를 찾을 수가 없어요..

KTable이 로컬 기반의 rocksDB에 기록된 데이터를 가져다 쓴다는 설명을 듣고 궁금증이 생깁니다.카프카 스트림을 docker기반 컨테이너로 배포한다 치면 그때마다 rocksDB는 초기화 될 것으로 생각되는데요..그렇다면 조인시에 데이터가 없어서 의도한 동작이 이루어지지 않을 수 있겠다는 생각이 드는데 맞는지요?마찬가지로 windowing 프로세스도 집계중에 컨테이너가 갈아쳐지면 집계중이던 데이터는 소실될 수 있어보이는데 맞는지 모르겠습니다.실무에서는 보통 카프카 스트림즈를 어떻게 쓰는지가 궁금합니다. rocksDB를 따로 두어서 모든 스트림이 공유하도록 설정하거나 컨테이너 배포는 사용하지 않는지.. 아니면 EC2같은곳에 배포해서 로컬 DB데이터를 훼손시키지 않도록 하는지 같은것이요.감사합니다.

- 본 강의 영상 학습 관련 문의에 대해 답변을 드립니다. (어떤 챕터 몇분 몇초를 꼭 기재부탁드립니다)- 이외의 문의등은 평생강의이므로 양해를 부탁드립니다- 현업과 병행하는 관계로 주말/휴가 제외 최대한 3일내로 답변을 드리려 노력하고 있습니다- 잠깐! 인프런 서비스 운영 관련 문의는 1:1 문의하기를 이용해주세요. 주피터 노트북 설치중인데 아나콘다에 없네요..ㅜㅜ어떻게 해야 할까요?

강의자님 강의자료를 통해서 노트에 학습 내용을 개인적으로 요약해서 개인 블로그에 공부 기록을 목적으로 사용하고 싶은데 가능할까요?

id(id는 auto increment 설정)와 name 등의 데이터를 5개를 추가한 후, delete from 을 통해 삭제하고 다시 데이터를 넣으니, id가 1부터 다시 지정되는 것이아닌 6부터 지정이 되더라구요..삭제 후, 다시 id를 1부터 하게 할 순 없을까요..?

만약 KStream, KTable 파티션 개수가 2개이고, 파티셔너 전략도 동일합니다. 근데 데이터 발생양이 증가하여, 파티션 개수를 둘다 5개를 늘려야 하는 상황이 생겼습니다.이때는 어떻게 해야하나요? 하나 씩 파티션을 증가할 때, 파티션 개수가 다르면 TopologyException이 발생할텐데요.또, 파티션이 추가되면 파티션 1번으로 가던 메시지 키가 다시 1번으로 간다는 보장도 없고요..2개의 토픽을 리파티셔닝 작업을 해야하는걸까요?리파티셔닝 작업을 하는 동안은 스트림즈의 다운 타임이 발생할 수 있는거고요..?

안녕하세요.강사님의 강의를 듣고, 이제는 강사님 책을 보고 있습니다. 실제 상용에서 애플리케이션을 개발할 때 궁금한 점이 있습니다.  첫번째로 프로듀서 애플리케이션 입니다.강사님의 책에서 봤듯이, 스프링을 사용합니다.예를 들어 저는 사용자 서비스에서 디비 트랜잭션(스프링에서 @Transactionl)을 사용하여 사용자 디비에 저장을 성공 후 프로듀서를 사용하여 레코드를 브로커에 보내야 한다고 생각합니다.밑에 코드를 간략하게 작성하였는데, 저렇게 되면 디비는 롤백됬지만 토픽에 레코드가 저장된 상태가 될 수 있다고 생각합니다.혹시 실무에서 커밋이 성공 후에 프로듀서 레코드를 전송하는 방법을 어떻게 하는지 간략한 코드가 궁금합니다.개인적인 생각은 UserService 클래스 상위 클래스로 카프카 프로듀서 처리하는 클래스를 만들어서 처리를 하는건지.. 궁금합니다.(이렇게 되면 카프카를 위한 래퍼 클래스가 항상 생기게 되는 불편함이 있는 것 같기도 하고요.. // 제가 생각하기에 잘못된 방법? // 만약 레코드 전송 후 어떤 이유로 에러가 발생하면, 디비에 저장된 데이터를 롤백되지만 // 프로듀서 레코드는 브로커 전송이 되버림 public class UserService { @Transactionl public void save(){ ... 프로듀서 레코드 전송 코드 ... 예외 발생 } } ----------------------- // 개인적인 생각 public class KafkaServce { public void save(){ userService.save(); 프로듀서 레코드 전송 코드 } } public class UserService { @Transactionl public void save(){ ... } }  두번째로 컨슈머 애플리케이션 개발 시 궁금한 점이 있습니다.컨슈머에서 데이터 처리를하다가 어떤 이유로 에러가 발생하여 해당 레코드 처리를 계속 실패했다고 가정합니다.그럴 경우 이 레코드의 대한 커밋 처리를 어떻게 해야할지가 궁금합니다.커밋을 처리하지 않으면, 다음 레코드 처리를 하지 못하는거라 생각되는데.. 어떤 방법으로 풀어내는지가 궁금합니다. 세번째로 컨슈머 애플리케이션에서 데이터베이스의 데이터를 저장해야하는 상황이다.스프링을 사용하는 경우 기본적으로 히카리 커넥션 풀에 커넥션 10개를 생성합니다.만약 파티션이 10개여서, 컨슈머를 10개 실행해야 한다면, 스프링 커넥션 풀을 사용하면 100개의 커넥션이 연결됩니다.50개면 500개의 풀 계속 증가할 듯 싶네요.이 경우 어떻게 해야할까요?컨슈머에서 레코드들을 for 문으로 돌리기 때문에 커넥션 풀을 1개를 사용해서 개발하는게 맞는건지?아니면 스프링에서 제공하는 히카리 커넥션 풀을 사용하지 않고, 직접 커넥션 풀을 구현하든가, 그것도 아니면 커넥션 풀을 사용하지 않고 1개의 트랜잭션당 1개의 커넥션을 생성 후 해제를 해야할까요?강사님의 생각이 궁금합니다. 마지막으로 궁금한게 있습니다.혹시 카프카를 활용하여 MSA에서 보상 트랜잭션(사가 패턴 - Orchestration) 처리를 할 수 있는건지 궁금합니다.

설치를 다 했는데, 다시 시작하면 초기화가 되어 있습니다.혹시 우분투 버전이 너무 낮아서 그런건지 / 저장하고 종료하고 싶은데 저장 및 종료가 잘 안되네요. 다 날아가서 당황스럽습니다.VMWARE에서 하는게 제일 안전한 것인지요 답변해주시면 감사하겠습니다 ~~

안녕하세요 본 강의 챕터9 마지막 강의와 챕터10 첫번쨰 강의가 재생되지 않습니다.지금까지 강의 잘 듣고있었는데 해당 부분만 진행이되지 않아 문의드립니다.

예전에 airflow를 쓰면서 DAG의 task가 스케쥴에 따라서 실행되지 않았던 적이 있습니다. 실행될 시간인데 scheduled상태로도 들어가지 않고 아예 사라졌다가 한참 후에 다시 scheduled되어서 중간에 실행되지 않는 시간이 많이 생겼습니다. 당시에는 pool의 slot수를 늘리니 해결이 되긴 했습니다. 그런데 강의에 따르면 slot수에 scheduled는 영향을 주지 않는 것으로 이해가 됐습니다.혹시 scheduled도 따로 slot개수를 가지게 되는건가요? p.s. 그때는 몰랐지만 airflow가 너무 자주 실행하는 경우에는 적합하지 않다는 거는 알게된 경험이었습니다. 당시 스케쥴은 1분마다 실행시키는 DAG들을 많이 만들었거든요

안녕하세요 강사님 좋은 강의 올려주셔서 감사합니다.강의 수강을 위해 원본 도표강의 자료 항목에 있는 강의 자료 다운로드 후 열어보려고하니 파일이 확인되지않습니다. (확장자 등을 변경해서 확인하려고해도 불러오지 못하고있습니다.) 혹시 별도의 pdf 리더가 필요한건가요? 바쁘시겟지만 확인 한번 부탁드립니다.

안녕하세요 선생님 🙂 PostgresOperator 질문 드립니다. DB table에 데이터를 갱신하는 task를 혼자 만들어보고 있는데요.PostgresOperator는 executemany와 같은 기능을 지원하지 않는 것으로 확인했습니다. airflow에서 대량의 데이터를 insert / update 하는 방법이 있을까요..?

Window에서 Volume을 적용할 때 cmd 환경에서 docker run -d -p 5000:8080 -v /usr/src/app/node_modules -v %cd%:/usr/src/app anonymous123/nodejs 해보고 wsl에서 docker run -d -p 5000:8080 -v /usr/src/app/node_modules -v "($pwd):/usr/src/app" anonymous123/nodejs를 시도하였는데 둘다 리액트가 작동은 하나 실시간 수정이 제대로 되지 않습니다 해결방법 알려주시면 감사하겠습니다 ㅜㅜ  

선생님 완강 이후 커뮤니티 이벤트 참여 신청했는데요이벤트 끝난건가요? ㅠ https://www.inflearn.com/course/%EC%8B%A4%EB%A6%AC%EC%BD%98%EB%B0%B8%EB%A6%AC-%EC%97%94%EC%A7%80%EB%8B%88%EC%96%B4%EC%99%80-%ED%95%A8%EA%BB%98%ED%95%98%EB%8A%94-apache-airflow/news

선생님 안녕하세요 좋은 강의 감사합니다. 2-4 강의에서 select fruit 하는 부분에서 오류가 생겨서 질문 드립니다. 참고로 저는 리눅스가 아니라 docker on window를 사용하고 있습니다.윈도우 상에서 bash 를 사용하는데 오류가 발생하는 것 같은데 해결방법을 모르겠어서요 저의 select_fruit.sh 는 아래와 같고요FRUIT=$1if [ $FRUIT == APPLE ];then echo "You selected Apple!"elif [ $FRUIT == ORANGE ];then echo "you selected Orange!"elif [ $FRUIT == GRAPE ];then echo "You selected Grape!"else echo "You selected other fruits!"fi 이것을 git bash 상에서 돌려보면 잘 돌아갑니다. 그런데, 윈도우 cmd에서 bash 로 돌리면 apple과 orange에 대해서도 다른 과일을 골랐다는 오류가 뜨고요 그리고 airflow dag을 실행시켜도 같은 오류가 뜹니다. git bash에서 제대로 결과가 나오는걸 보면 sh 파일 자체에는 문제가 없는것 같은데, airflow 상에서 실행이 안되는 이유가 무엇일까요 ㅠ

지금 강의 소스 코드가 제대로 다운로드 되지 않습니다 문제 해결해주시면 감사하겠습니다